直接发im，去掉 outbox一层
This commit is contained in:
zhx
parent
2e061101b3
commit
a1e038442d
File diff suppressed because it is too large
Load Diff
@ -1067,6 +1067,10 @@ message SendGiftResponse {
|
||||
LuckyGiftDrawResult lucky_gift = 7;
|
||||
// lucky_gifts 是每个收礼目标的独立抽奖结果；多目标客户端用 target_user_id 对齐明细表现。
|
||||
repeated LuckyGiftDrawResult lucky_gifts = 8;
|
||||
// coin_balance_after 是送礼人本次扣费和幸运礼物返奖完成后的 COIN 可用余额，只返回给送礼发起方。
|
||||
int64 coin_balance_after = 9;
|
||||
// gift_income_balance_after 是收礼人返币后的 COIN 余额；多目标或无返币时为 0。
|
||||
int64 gift_income_balance_after = 10;
|
||||
}
|
||||
|
||||
// CheckSpeakPermissionRequest 让腾讯云 IM 发言回调或 gateway 在公屏前同步问房间业务态。
|
||||
|
||||
@ -3,7 +3,7 @@ package room
|
||||
import "context"
|
||||
|
||||
// EventHandler 是 room outbox 事件进入 activity-service 后的处理边界。
|
||||
// 消费来源可以是 MySQL outbox、Redis Stream 或 MQ，但处理侧必须保持幂等。
|
||||
// 消费来源可以是 MySQL outbox 或 MQ，但处理侧必须保持幂等。
|
||||
type EventHandler interface {
|
||||
HandleRoomEvent(ctx context.Context, eventID string, eventType string, payload []byte) error
|
||||
}
|
||||
|
||||
@ -2597,6 +2597,7 @@ func TestSendGiftResponseIncludesLuckyGiftDraw(t *testing.T) {
|
||||
Result: &roomv1.CommandResult{Applied: true, RoomVersion: 14, ServerTimeMs: 1_779_258_000_000},
|
||||
BillingReceiptId: "receipt-lucky",
|
||||
RoomHeat: 100,
|
||||
CoinBalanceAfter: 8800,
|
||||
LuckyGift: &roomv1.LuckyGiftDrawResult{
|
||||
Enabled: true,
|
||||
DrawId: "lucky_draw_test",
|
||||
@ -2644,6 +2645,7 @@ func TestSendGiftResponseIncludesLuckyGiftDraw(t *testing.T) {
|
||||
MultiplierPPM int64 `json:"multiplier_ppm"`
|
||||
EffectiveRewardCoins int64 `json:"effective_reward_coins"`
|
||||
} `json:"lucky_gifts"`
|
||||
CoinBalanceAfter int64 `json:"coin_balance_after"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
|
||||
@ -2655,6 +2657,9 @@ func TestSendGiftResponseIncludesLuckyGiftDraw(t *testing.T) {
|
||||
if len(envelope.Data.LuckyGifts) != 2 || envelope.Data.LuckyGifts[0].TargetUserID != 43 || envelope.Data.LuckyGifts[0].MultiplierPPM != 2_000_000 || envelope.Data.LuckyGifts[1].TargetUserID != 44 || envelope.Data.LuckyGifts[1].MultiplierPPM != 3_000_000 {
|
||||
t.Fatalf("lucky_gifts response mismatch: %+v", envelope.Data.LuckyGifts)
|
||||
}
|
||||
if envelope.Data.CoinBalanceAfter != 8800 {
|
||||
t.Fatalf("send gift response must expose sender coin_balance_after: %+v", envelope.Data)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSendGiftInjectsActiveHostPeriodScope(t *testing.T) {
|
||||
|
||||
33
services/room-service/README.md
Normal file
33
services/room-service/README.md
Normal file
@ -0,0 +1,33 @@
|
||||
# room-service
|
||||
|
||||
`room-service` owns Room Cell state, room snapshots, command log, durable room facts, Redis lease, and Tencent IM room message bridging.
|
||||
|
||||
## Direct IM UI
|
||||
|
||||
Room UI events are sent to Tencent IM directly after the Room Cell command commits. The direct IM lane is low latency and best-effort; it is not the owner of recoverable room state.
|
||||
|
||||
Direct-IM-only event types do not write `room_outbox` because snapshot or query APIs can repair missed display state:
|
||||
|
||||
- `RoomHeatChanged`: heat display; snapshot can repair missed values.
|
||||
- `RoomRankChanged`: mic/rank display; direct IM display is preferred.
|
||||
- `RoomRocketFuelChanged`: coalesced in memory and flushed directly to Tencent IM.
|
||||
- `RoomRocketIgnited`: sent directly to Tencent IM; launch scheduling is still created after command commit.
|
||||
- `RoomRocketLaunched`: launch display is sent directly to Tencent IM.
|
||||
|
||||
Durable facts with UI display are sent directly to Tencent IM after the Room Cell command commits, but they still write `room_outbox`:
|
||||
|
||||
- `RoomGiftSent`: room gift animation, public screen, and regional display use direct IM; statistics, activity, growth, task, and CP relation consumers still use durable outbox.
|
||||
- `RoomUserJoined`: entry animation, vehicle, and online display use direct IM; active-user statistics still use durable outbox.
|
||||
- `RoomUserLeft`: leave UI uses direct IM; recovery still comes from Room Cell command log and snapshot.
|
||||
- `RoomMicChanged`: mic UI uses direct IM; mic duration consumers still use durable outbox.
|
||||
- `RoomMicSeatLocked`, `RoomClosed`, `RoomUserKicked`, `RoomUserMuted`, `RoomUserUnbanned`, `RoomProfileUpdated`, `RoomBackgroundChanged`, `RoomChatEnabledChanged`, and `RoomAdminChanged`: UI updates use direct IM; state changes, side effects, audit, or notification recovery stay on command log/outbox.
|
||||
|
||||
## Durable Outbox
|
||||
|
||||
Business facts that downstream services must not miss still write `room_outbox` and are published by the outbox worker. Sending a display event directly to Tencent IM does not change command log, snapshot, room lease, wallet debit, or room state ownership.
|
||||
|
||||
The room outbox worker no longer publishes room UI IM. It only fans out durable facts to MQ/activity consumers. Gift sender balance is not included in room IM because room messages are broadcast to all room users. `SendGiftResponse.coin_balance_after` returns the sender's final COIN balance synchronously, including lucky-gift reward balance when a reward is granted.
|
||||
|
||||
`RoomRocketIgnited` keeps a reliability check outside direct IM: after the room command commits, room-service still schedules the delayed launch wakeup, and the launch worker revalidates rocket id, level, launch time, status, reset time, and room lease before creating `RoomRocketLaunched`.
|
||||
|
||||
Robot display events keep their original best-effort display publisher path in this pass.
|
||||
@ -67,11 +67,6 @@ rocketmq:
|
||||
enabled: true
|
||||
topic: "hyapp_room_outbox"
|
||||
producer_group: "hyapp-room-outbox-producer"
|
||||
tencent_im_consumer_enabled: true
|
||||
tencent_im_consumer_group: "hyapp-room-im-bridge"
|
||||
consumer_max_reconsume_times: 16
|
||||
consumer_goroutines: 16
|
||||
im_publish_concurrency: 16
|
||||
rocket_launch:
|
||||
enabled: true
|
||||
topic: "hyapp_room_rocket_launch"
|
||||
|
||||
@ -77,11 +77,6 @@ rocketmq:
|
||||
enabled: true
|
||||
topic: "hyapp_room_outbox"
|
||||
producer_group: "hyapp-room-outbox-producer"
|
||||
tencent_im_consumer_enabled: true
|
||||
tencent_im_consumer_group: "hyapp-room-im-bridge"
|
||||
consumer_max_reconsume_times: 16
|
||||
consumer_goroutines: 16
|
||||
im_publish_concurrency: 16
|
||||
rocket_launch:
|
||||
enabled: true
|
||||
topic: "hyapp_room_rocket_launch"
|
||||
|
||||
@ -70,11 +70,6 @@ rocketmq:
|
||||
enabled: true
|
||||
topic: "hyapp_room_outbox"
|
||||
producer_group: "hyapp-room-outbox-producer"
|
||||
tencent_im_consumer_enabled: true
|
||||
tencent_im_consumer_group: "hyapp-room-im-bridge"
|
||||
consumer_max_reconsume_times: 16
|
||||
consumer_goroutines: 16
|
||||
im_publish_concurrency: 16
|
||||
rocket_launch:
|
||||
enabled: true
|
||||
topic: "hyapp_room_rocket_launch"
|
||||
|
||||
@ -13,7 +13,6 @@ import (
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
activityv1 "hyapp.local/api/proto/activity/v1"
|
||||
roomv1 "hyapp.local/api/proto/room/v1"
|
||||
"hyapp/pkg/appcode"
|
||||
"hyapp/pkg/healthhttp"
|
||||
"hyapp/pkg/logx"
|
||||
"hyapp/pkg/rocketmqx"
|
||||
@ -182,9 +181,6 @@ func New(cfg config.Config) (*App, error) {
|
||||
activityPublisher := integration.NewActivityOutboxPublisher(activityv1.NewRoomEventConsumerServiceClient(activityConn), time.Second)
|
||||
outboxPublishers = append(outboxPublishers, activityPublisher)
|
||||
}
|
||||
if tencentPublisher != nil {
|
||||
outboxPublishers = append(outboxPublishers, tencentPublisher)
|
||||
}
|
||||
}
|
||||
if cfg.OutboxWorker.PublishMode == config.OutboxPublishModeMQ || cfg.OutboxWorker.PublishMode == config.OutboxPublishModeDual {
|
||||
outboxProducer, err := rocketmqx.NewProducer(rocketMQProducerConfig(cfg.RocketMQ, cfg.RocketMQ.RoomOutbox.ProducerGroup))
|
||||
@ -227,6 +223,7 @@ func New(cfg config.Config) (*App, error) {
|
||||
RTCUserRemover: rtcUserRemover,
|
||||
RoomRocketLaunchScheduler: rocketLaunchScheduler,
|
||||
RobotDisplayPublisher: tencentPublisher,
|
||||
RoomDisplayPublisher: tencentPublisher,
|
||||
HumanRobotPoolProvider: integration.NewGRPCHumanRoomRobotPoolProvider(robotConn, userConn),
|
||||
RoomGiftLeaderboard: roomservice.NewRedisRoomGiftLeaderboardStore(redisClient),
|
||||
RoomListCache: roomservice.NewRedisRoomListCacheStore(redisClient),
|
||||
@ -257,38 +254,6 @@ func New(cfg config.Config) (*App, error) {
|
||||
}
|
||||
mqConsumers = append(mqConsumers, rocketConsumer)
|
||||
}
|
||||
if cfg.RocketMQ.RoomOutbox.TencentIMConsumerEnabled && tencentPublisher == nil {
|
||||
_ = repository.Close()
|
||||
closeClientConn(activityConn)
|
||||
_ = walletConn.Close()
|
||||
_ = redisClient.Close()
|
||||
return nil, errors.New("rocketmq room_outbox tencent_im consumer requires tencent_im.enabled")
|
||||
}
|
||||
if cfg.RocketMQ.RoomOutbox.TencentIMConsumerEnabled && tencentPublisher != nil {
|
||||
imPublisher := integration.NewShardedOutboxPublisher(tencentPublisher, cfg.RocketMQ.RoomOutbox.IMPublishConcurrency)
|
||||
imConsumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.RoomOutbox.TencentIMConsumerGroup, cfg.RocketMQ.RoomOutbox.ConsumerMaxReconsumeTimes, cfg.RocketMQ.RoomOutbox.ConsumerGoroutines))
|
||||
if err != nil {
|
||||
_ = repository.Close()
|
||||
closeClientConn(activityConn)
|
||||
_ = walletConn.Close()
|
||||
_ = redisClient.Close()
|
||||
return nil, err
|
||||
}
|
||||
if err := imConsumer.Subscribe(cfg.RocketMQ.RoomOutbox.Topic, roommq.TagRoomOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
||||
envelope, _, err := roommq.DecodeRoomOutboxMessage(message.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return imPublisher.PublishOutboxEvent(appcode.WithContext(ctx, envelope.GetAppCode()), envelope)
|
||||
}); err != nil {
|
||||
_ = repository.Close()
|
||||
closeClientConn(activityConn)
|
||||
_ = walletConn.Close()
|
||||
_ = redisClient.Close()
|
||||
return nil, err
|
||||
}
|
||||
mqConsumers = append(mqConsumers, imConsumer)
|
||||
}
|
||||
if cfg.RocketMQ.UserOutbox.Enabled {
|
||||
userConsumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.UserOutbox.ConsumerGroup, cfg.RocketMQ.UserOutbox.ConsumerMaxReconsumeTimes, 0))
|
||||
if err != nil {
|
||||
|
||||
@ -137,15 +137,16 @@ type RocketMQConfig struct {
|
||||
|
||||
// RoomOutboxMQConfig 控制 room_outbox fanout topic。
|
||||
type RoomOutboxMQConfig struct {
|
||||
Enabled bool `yaml:"enabled"`
|
||||
Topic string `yaml:"topic"`
|
||||
ProducerGroup string `yaml:"producer_group"`
|
||||
Enabled bool `yaml:"enabled"`
|
||||
Topic string `yaml:"topic"`
|
||||
ProducerGroup string `yaml:"producer_group"`
|
||||
// TencentIMConsumerEnabled 已废弃：房间 UI 消息由 Room Cell 提交后直发腾讯 IM，room_outbox 只保留业务事实 fanout。
|
||||
TencentIMConsumerEnabled bool `yaml:"tencent_im_consumer_enabled"`
|
||||
TencentIMConsumerGroup string `yaml:"tencent_im_consumer_group"`
|
||||
ConsumerMaxReconsumeTimes int32 `yaml:"consumer_max_reconsume_times"`
|
||||
// ConsumerGoroutines 控制 RocketMQ push consumer 的并发回调数；同房间顺序仍由 IM publish 分片锁保证。
|
||||
// ConsumerGoroutines 控制 room_outbox RocketMQ push consumer 的并发回调数，消费方仍必须自行做业务幂等。
|
||||
ConsumerGoroutines int `yaml:"consumer_goroutines"`
|
||||
// IMPublishConcurrency 控制 MQ -> Tencent IM 桥接按 room_id 分片后的最大并行房间数。
|
||||
// IMPublishConcurrency 已废弃，仅兼容旧配置文件；房间 IM 不再由 MQ consumer 桥接。
|
||||
IMPublishConcurrency int `yaml:"im_publish_concurrency"`
|
||||
}
|
||||
|
||||
@ -321,7 +322,7 @@ func defaultRocketMQConfig() RocketMQConfig {
|
||||
Topic: "hyapp_room_outbox",
|
||||
ProducerGroup: "hyapp-room-outbox-producer",
|
||||
TencentIMConsumerEnabled: false,
|
||||
TencentIMConsumerGroup: "hyapp-room-im-bridge",
|
||||
TencentIMConsumerGroup: "",
|
||||
ConsumerMaxReconsumeTimes: 16,
|
||||
IMPublishConcurrency: 1,
|
||||
},
|
||||
@ -519,6 +520,8 @@ func normalizeRocketMQConfig(cfg RocketMQConfig, publishMode string) (RocketMQCo
|
||||
if cfg.RoomOutbox.ConsumerMaxReconsumeTimes <= 0 {
|
||||
cfg.RoomOutbox.ConsumerMaxReconsumeTimes = defaults.RoomOutbox.ConsumerMaxReconsumeTimes
|
||||
}
|
||||
// 旧配置可能仍把 MQ -> Tencent IM bridge 打开；当前设计要求 UI 在 Room Cell 提交后直接 IM，outbox worker 不再承担 IM 投递。
|
||||
cfg.RoomOutbox.TencentIMConsumerEnabled = false
|
||||
if cfg.RoomOutbox.ConsumerGoroutines < 0 {
|
||||
cfg.RoomOutbox.ConsumerGoroutines = 0
|
||||
}
|
||||
@ -546,7 +549,7 @@ func normalizeRocketMQConfig(cfg RocketMQConfig, publishMode string) (RocketMQCo
|
||||
if cfg.UserOutbox.ConsumerMaxReconsumeTimes <= 0 {
|
||||
cfg.UserOutbox.ConsumerMaxReconsumeTimes = defaults.UserOutbox.ConsumerMaxReconsumeTimes
|
||||
}
|
||||
if cfg.RoomOutbox.Enabled || cfg.RocketLaunch.Enabled || cfg.UserOutbox.Enabled || cfg.RoomOutbox.TencentIMConsumerEnabled {
|
||||
if cfg.RoomOutbox.Enabled || cfg.RocketLaunch.Enabled || cfg.UserOutbox.Enabled {
|
||||
cfg.Enabled = true
|
||||
}
|
||||
if publishMode == OutboxPublishModeMQ || publishMode == OutboxPublishModeDual {
|
||||
|
||||
@ -43,12 +43,9 @@ func TestLoad(t *testing.T) {
|
||||
if !cfg.OutboxWorker.Enabled || cfg.OutboxWorker.PublishMode != OutboxPublishModeMQ || cfg.OutboxWorker.PollInterval != 200*time.Millisecond || cfg.OutboxWorker.BatchSize != 50 || cfg.OutboxWorker.Concurrency != 16 || cfg.OutboxWorker.PublishTimeout != 15*time.Second || cfg.OutboxWorker.RetryStrategy != "exponential_backoff" || cfg.OutboxWorker.MaxRetryCount != 10 || cfg.OutboxWorker.InitialBackoff != 5*time.Second || cfg.OutboxWorker.MaxBackoff != 5*time.Minute {
|
||||
t.Fatalf("unexpected outbox worker config: %+v", cfg.OutboxWorker)
|
||||
}
|
||||
// 本地默认也走 room_outbox MQ、IM bridge 和火箭延迟唤醒，避免本地直投和线上 MQ 语义分叉。
|
||||
if !cfg.RocketMQ.Enabled || !cfg.RocketMQ.RoomOutbox.Enabled || !cfg.RocketMQ.RoomOutbox.TencentIMConsumerEnabled || !cfg.RocketMQ.RocketLaunch.Enabled || !cfg.RocketMQ.UserOutbox.Enabled {
|
||||
t.Fatalf("local config must enable room MQ fanout and IM bridge: %+v", cfg.RocketMQ)
|
||||
}
|
||||
if cfg.RocketMQ.RoomOutbox.ConsumerGoroutines != 16 || cfg.RocketMQ.RoomOutbox.IMPublishConcurrency != 16 {
|
||||
t.Fatalf("local IM bridge concurrency mismatch: room=%+v", cfg.RocketMQ.RoomOutbox)
|
||||
// 本地默认仍走 room_outbox MQ 和火箭延迟唤醒；房间 UI IM 已改为 Room Cell 提交后直发，不再启 MQ IM bridge。
|
||||
if !cfg.RocketMQ.Enabled || !cfg.RocketMQ.RoomOutbox.Enabled || cfg.RocketMQ.RoomOutbox.TencentIMConsumerEnabled || !cfg.RocketMQ.RocketLaunch.Enabled || !cfg.RocketMQ.UserOutbox.Enabled {
|
||||
t.Fatalf("local config must enable room MQ fanout and disable MQ IM bridge: %+v", cfg.RocketMQ)
|
||||
}
|
||||
}
|
||||
|
||||
@ -60,7 +57,7 @@ func TestLoadTencentExample(t *testing.T) {
|
||||
}
|
||||
|
||||
if !cfg.TencentIM.Enabled {
|
||||
t.Fatalf("tencent example should enable Tencent IM bridge")
|
||||
t.Fatalf("tencent example should enable Tencent IM direct publisher")
|
||||
}
|
||||
if cfg.TencentIM.SDKAppID == 0 || cfg.TencentIM.SecretKey == "" {
|
||||
t.Fatalf("sdk_app_id and secret_key templates must be present")
|
||||
@ -86,12 +83,9 @@ func TestLoadTencentExample(t *testing.T) {
|
||||
if cfg.OutboxWorker.PublishMode != OutboxPublishModeMQ || !cfg.RocketMQ.RoomOutbox.Enabled || !cfg.RocketMQ.RocketLaunch.Enabled {
|
||||
t.Fatalf("tencent example must route room outbox and rocket launch through RocketMQ: mode=%s mq=%+v", cfg.OutboxWorker.PublishMode, cfg.RocketMQ)
|
||||
}
|
||||
if cfg.RocketMQ.RoomOutbox.ConsumerGoroutines != 16 || cfg.RocketMQ.RoomOutbox.IMPublishConcurrency != 16 {
|
||||
t.Fatalf("prod IM bridge concurrency mismatch: room=%+v", cfg.RocketMQ.RoomOutbox)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadDockerConfigStartsRoomIMBridgeWhenOutboxUsesMQ(t *testing.T) {
|
||||
func TestLoadDockerConfigDoesNotStartRoomIMBridgeWhenOutboxUsesMQ(t *testing.T) {
|
||||
cfg, err := Load("../../configs/config.docker.yaml")
|
||||
if err != nil {
|
||||
t.Fatalf("Load docker config failed: %v", err)
|
||||
@ -106,17 +100,14 @@ func TestLoadDockerConfigStartsRoomIMBridgeWhenOutboxUsesMQ(t *testing.T) {
|
||||
if !cfg.RocketMQ.RoomOutbox.Enabled {
|
||||
t.Fatalf("docker config must enable room outbox MQ")
|
||||
}
|
||||
if !cfg.RocketMQ.RoomOutbox.TencentIMConsumerEnabled {
|
||||
t.Fatalf("docker mq room outbox must start Tencent IM bridge consumer")
|
||||
}
|
||||
if cfg.RocketMQ.RoomOutbox.ConsumerGoroutines != 16 || cfg.RocketMQ.RoomOutbox.IMPublishConcurrency != 16 {
|
||||
t.Fatalf("docker room IM bridge concurrency mismatch: %+v", cfg.RocketMQ.RoomOutbox)
|
||||
if cfg.RocketMQ.RoomOutbox.TencentIMConsumerEnabled {
|
||||
t.Fatalf("docker mq room outbox must not start Tencent IM bridge consumer")
|
||||
}
|
||||
if cfg.TencentIM.SDKAppID != 20040101 || cfg.TencentIM.AdminIdentifier != "administrator" {
|
||||
t.Fatalf("docker config must use test IM account: im=%+v", cfg.TencentIM)
|
||||
}
|
||||
if !cfg.TencentIM.Enabled {
|
||||
t.Fatalf("Tencent IM bridge consumer requires tencent_im.enabled")
|
||||
t.Fatalf("direct room IM publishing requires tencent_im.enabled")
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
117
services/room-service/internal/room/service/direct_im.go
Normal file
117
services/room-service/internal/room/service/direct_im.go
Normal file
@ -0,0 +1,117 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"hyapp/pkg/appcode"
|
||||
"hyapp/pkg/logx"
|
||||
"hyapp/services/room-service/internal/room/outbox"
|
||||
)
|
||||
|
||||
const roomDirectIMPublishTimeout = 2 * time.Second
|
||||
|
||||
func splitRoomDirectIMRecords(records []outbox.Record) ([]outbox.Record, []outbox.Record) {
|
||||
if len(records) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
durable := make([]outbox.Record, 0, len(records))
|
||||
directIM := make([]outbox.Record, 0)
|
||||
for _, record := range records {
|
||||
if roomDirectIMOnlyEventType(record.EventType) {
|
||||
directIM = append(directIM, record)
|
||||
continue
|
||||
}
|
||||
durable = append(durable, record)
|
||||
if roomDirectIMMirrorEventType(record.EventType) {
|
||||
// 这类事件同时服务 UI 和业务事实消费：MySQL outbox 保证统计、任务、通知等可补偿，直接 IM 只给客户端低延迟展示。
|
||||
directIM = append(directIM, record)
|
||||
}
|
||||
}
|
||||
return durable, directIM
|
||||
}
|
||||
|
||||
func roomDirectIMOnlyEventType(eventType string) bool {
|
||||
switch eventType {
|
||||
case "RoomHeatChanged",
|
||||
"RoomRankChanged",
|
||||
"RoomRocketFuelChanged",
|
||||
"RoomRocketLaunched":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func roomDirectIMMirrorEventType(eventType string) bool {
|
||||
switch eventType {
|
||||
case "RoomGiftSent",
|
||||
"RoomUserJoined",
|
||||
"RoomUserLeft",
|
||||
"RoomMicChanged",
|
||||
"RoomMicSeatLocked",
|
||||
"RoomClosed",
|
||||
"RoomUserKicked",
|
||||
"RoomUserMuted",
|
||||
"RoomUserUnbanned",
|
||||
"RoomProfileUpdated",
|
||||
"RoomBackgroundChanged",
|
||||
"RoomChatEnabledChanged",
|
||||
"RoomAdminChanged",
|
||||
"RoomRocketIgnited":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) publishDirectIMRecordsBestEffort(ctx context.Context, records []outbox.Record) {
|
||||
if s == nil || len(records) == 0 {
|
||||
return
|
||||
}
|
||||
if s.roomDisplayPublisher == nil {
|
||||
logx.Warn(ctx, "room_direct_im_publish_skipped",
|
||||
slog.String("node_id", s.nodeID),
|
||||
slog.Int("record_count", len(records)),
|
||||
slog.String("reason", "publisher_not_configured"),
|
||||
)
|
||||
return
|
||||
}
|
||||
scopedApp := appcode.FromContext(ctx)
|
||||
copied := append([]outbox.Record(nil), records...)
|
||||
for index := range copied {
|
||||
copied[index].AppCode = appcode.Normalize(firstNonEmpty(copied[index].AppCode, scopedApp))
|
||||
if copied[index].Envelope != nil {
|
||||
copied[index].Envelope.AppCode = copied[index].AppCode
|
||||
}
|
||||
}
|
||||
sortOutboxRecords(copied)
|
||||
|
||||
go func(records []outbox.Record) {
|
||||
for _, record := range records {
|
||||
if record.Envelope == nil {
|
||||
logx.Warn(ctx, "room_direct_im_publish_skipped",
|
||||
slog.String("node_id", s.nodeID),
|
||||
slog.String("event_id", record.EventID),
|
||||
slog.String("event_type", record.EventType),
|
||||
slog.String("room_id", record.RoomID),
|
||||
slog.String("reason", "nil_envelope"),
|
||||
)
|
||||
continue
|
||||
}
|
||||
publishCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), roomDirectIMPublishTimeout)
|
||||
err := s.roomDisplayPublisher.PublishOutboxEvent(publishCtx, record.Envelope)
|
||||
cancel()
|
||||
if err != nil {
|
||||
logx.Warn(ctx, "room_direct_im_publish_failed",
|
||||
slog.String("node_id", s.nodeID),
|
||||
slog.String("event_id", record.EventID),
|
||||
slog.String("event_type", record.EventType),
|
||||
slog.String("room_id", record.RoomID),
|
||||
slog.String("error", trimOutboxError(err.Error())),
|
||||
)
|
||||
}
|
||||
}
|
||||
}(copied)
|
||||
}
|
||||
@ -388,18 +388,21 @@ func (s *Service) sendGift(ctx context.Context, req *roomv1.SendGiftRequest, rob
|
||||
}
|
||||
}
|
||||
|
||||
finalCoinBalanceAfter := giftFinalCoinBalanceAfter(billing, luckyGift)
|
||||
firstTargetDisplayProfile := giftDisplayProfileForUser(cmd.TargetDisplayProfiles, cmd.TargetUserID)
|
||||
result := mutationResult{
|
||||
snapshot: current.ToProto(),
|
||||
billingReceiptID: billing.GetBillingReceiptId(),
|
||||
roomHeat: current.Heat,
|
||||
giftRank: cloneProtoRank(current.GiftRank),
|
||||
luckyGift: luckyGift,
|
||||
luckyGifts: luckyGifts,
|
||||
commandPayload: commandPayload,
|
||||
walletDebitMS: walletDebitMS,
|
||||
robotDisplayRecords: robotDisplayRecords,
|
||||
roomRocketProgress: rocketProgress,
|
||||
snapshot: current.ToProto(),
|
||||
billingReceiptID: billing.GetBillingReceiptId(),
|
||||
coinBalanceAfter: finalCoinBalanceAfter,
|
||||
giftIncomeBalanceAfter: billing.GetGiftIncomeBalanceAfter(),
|
||||
roomHeat: current.Heat,
|
||||
giftRank: cloneProtoRank(current.GiftRank),
|
||||
luckyGift: luckyGift,
|
||||
luckyGifts: luckyGifts,
|
||||
commandPayload: commandPayload,
|
||||
walletDebitMS: walletDebitMS,
|
||||
robotDisplayRecords: robotDisplayRecords,
|
||||
roomRocketProgress: rocketProgress,
|
||||
roomGiftLeaderboard: &RoomGiftLeaderboardIncrement{
|
||||
AppCode: appcode.FromContext(ctx),
|
||||
RoomID: current.RoomID,
|
||||
@ -474,14 +477,16 @@ func (s *Service) sendGift(ctx context.Context, req *roomv1.SendGiftRequest, rob
|
||||
}
|
||||
|
||||
return &roomv1.SendGiftResponse{
|
||||
Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()),
|
||||
BillingReceiptId: result.billingReceiptID,
|
||||
RoomHeat: result.roomHeat,
|
||||
GiftRank: result.giftRank,
|
||||
Room: result.snapshot,
|
||||
Rocket: result.snapshot.GetRocket(),
|
||||
LuckyGift: result.luckyGift,
|
||||
LuckyGifts: result.luckyGifts,
|
||||
Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()),
|
||||
BillingReceiptId: result.billingReceiptID,
|
||||
RoomHeat: result.roomHeat,
|
||||
GiftRank: result.giftRank,
|
||||
Room: result.snapshot,
|
||||
Rocket: result.snapshot.GetRocket(),
|
||||
LuckyGift: result.luckyGift,
|
||||
LuckyGifts: result.luckyGifts,
|
||||
CoinBalanceAfter: result.coinBalanceAfter,
|
||||
GiftIncomeBalanceAfter: result.giftIncomeBalanceAfter,
|
||||
}, nil
|
||||
}
|
||||
|
||||
@ -766,6 +771,17 @@ func aggregateLuckyGiftResults(commandID string, results []*roomv1.LuckyGiftDraw
|
||||
return aggregate
|
||||
}
|
||||
|
||||
func giftFinalCoinBalanceAfter(billing *walletv1.DebitGiftResponse, luckyGift *roomv1.LuckyGiftDrawResult) int64 {
|
||||
if luckyGift != nil && luckyGift.GetCoinBalanceAfter() > 0 {
|
||||
// 幸运礼物返奖会在扣费后追加一笔入账；同步响应返回最终余额，避免客户端先显示扣费余额再等待中奖回写。
|
||||
return luckyGift.GetCoinBalanceAfter()
|
||||
}
|
||||
if billing == nil {
|
||||
return 0
|
||||
}
|
||||
return billing.GetBalanceAfter()
|
||||
}
|
||||
|
||||
func normalizeGiftTargetType(raw string) string {
|
||||
raw = strings.TrimSpace(raw)
|
||||
switch raw {
|
||||
|
||||
@ -326,7 +326,7 @@ func (s *Service) markOutboxDead(ctx context.Context, eventID string, lastErr st
|
||||
}
|
||||
|
||||
func (s *Service) publishOutboxRecord(ctx context.Context, record outbox.Record) error {
|
||||
// Ignited 先安排延迟发射，再对外广播事件；如果 MQ 延迟调度不可用，本条 outbox 保持 retryable。
|
||||
// Ignited 在 outbox worker 内安排延迟发射，保证 MQ 调度失败能通过 room_outbox retry；客户端 IM 展示已在 Room Cell 提交后直发。
|
||||
if err := s.scheduleRoomRocketLaunchFromEnvelope(ctx, record.Envelope); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@ -79,6 +79,7 @@ func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayabl
|
||||
}
|
||||
// 机器人展示只是 App 视觉补偿，提交前按短窗口降采样；命令日志、房间状态和主 outbox 不受影响。
|
||||
robotDisplayRecords := s.sampleRobotDisplayRecords(result.robotDisplayRecords)
|
||||
durableOutboxRecords, directIMRecords := splitRoomDirectIMRecords(outboxRecords)
|
||||
saveStartedAt := time.Now()
|
||||
if err := s.repository.SaveMutation(ctx, MutationCommit{
|
||||
Command: CommandRecord{
|
||||
@ -93,7 +94,7 @@ func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayabl
|
||||
LeaseToken: lease.LeaseToken,
|
||||
CreatedAtMS: now.UnixMilli(),
|
||||
},
|
||||
OutboxRecords: outboxRecords,
|
||||
OutboxRecords: durableOutboxRecords,
|
||||
RoomStatus: result.roomStatus,
|
||||
RoomSeatCount: result.roomSeatCount,
|
||||
RoomPasswordHash: result.roomPasswordHash,
|
||||
@ -106,11 +107,14 @@ func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayabl
|
||||
return nil, err
|
||||
}
|
||||
saveMutationMS += elapsedMS(saveStartedAt)
|
||||
// 保存已提交 outbox 记录，出 Cell 锁后由 outbox worker 异步投递 IM/activity，避免外部慢调用占住房间串行执行队列。
|
||||
result.outboxRecords = outboxRecords
|
||||
// 保存已提交 outbox 记录，出 Cell 锁后由 outbox worker 异步投递 activity/MQ 业务事实，避免外部慢调用占住房间串行执行队列。
|
||||
result.outboxRecords = durableOutboxRecords
|
||||
result.robotDisplayRecords = robotDisplayRecords
|
||||
result.directIMRecords = directIMRecords
|
||||
// MySQL 事务成功提交后再发本进程唤醒信号；信号丢失只影响延迟，不影响后续周期扫描补偿。
|
||||
s.notifyOutboxCommitted(outboxRecords)
|
||||
s.notifyOutboxCommitted(durableOutboxRecords)
|
||||
// 直接 IM lane 只承载展示事件；RoomRocketIgnited 同时保留 durable outbox，让延迟发射调度继续可重试。
|
||||
s.publishDirectIMRecordsBestEffort(ctx, directIMRecords)
|
||||
// 机器人展示事件允许极小概率丢失，不写 MySQL outbox；提交成功后后台直发 IM，失败只记录日志。
|
||||
s.publishRobotDisplayRecordsBestEffort(ctx, robotDisplayRecords)
|
||||
|
||||
|
||||
@ -8,6 +8,7 @@ import (
|
||||
|
||||
"hyapp/pkg/appcode"
|
||||
"hyapp/pkg/logx"
|
||||
"hyapp/services/room-service/internal/integration"
|
||||
"hyapp/services/room-service/internal/room/outbox"
|
||||
)
|
||||
|
||||
@ -45,15 +46,15 @@ type roomRocketProgressEntry struct {
|
||||
}
|
||||
|
||||
type roomRocketProgressCoalescer struct {
|
||||
mu sync.Mutex
|
||||
repository Repository
|
||||
pending map[roomRocketProgressKey]roomRocketProgressEntry
|
||||
mu sync.Mutex
|
||||
publisher integration.OutboxPublisher
|
||||
pending map[roomRocketProgressKey]roomRocketProgressEntry
|
||||
}
|
||||
|
||||
func newRoomRocketProgressCoalescer(repository Repository) *roomRocketProgressCoalescer {
|
||||
func newRoomRocketProgressCoalescer(publisher integration.OutboxPublisher) *roomRocketProgressCoalescer {
|
||||
return &roomRocketProgressCoalescer{
|
||||
repository: repository,
|
||||
pending: make(map[roomRocketProgressKey]roomRocketProgressEntry),
|
||||
publisher: publisher,
|
||||
pending: make(map[roomRocketProgressKey]roomRocketProgressEntry),
|
||||
}
|
||||
}
|
||||
|
||||
@ -95,7 +96,7 @@ func (c *roomRocketProgressCoalescer) enqueue(progress roomRocketProgressPending
|
||||
}
|
||||
|
||||
func (c *roomRocketProgressCoalescer) flushDue(ctx context.Context, nowMS int64) ([]outbox.Record, error) {
|
||||
if c == nil || c.repository == nil {
|
||||
if c == nil {
|
||||
return nil, nil
|
||||
}
|
||||
due := c.takeDue(nowMS)
|
||||
@ -103,8 +104,7 @@ func (c *roomRocketProgressCoalescer) flushDue(ctx context.Context, nowMS int64)
|
||||
return nil, nil
|
||||
}
|
||||
normalEntries, robotRecords := splitRoomRocketProgressDue(due)
|
||||
if err := c.saveNormalDue(ctx, normalEntries); err != nil {
|
||||
// 主 outbox 保存失败时只把主链路进度放回队列；机器人展示允许丢失，仍交给直发 IM best-effort 处理。
|
||||
if err := c.publishDue(ctx, normalEntries); err != nil {
|
||||
c.requeue(normalEntries, nowMS+roomRocketProgressFlushInterval.Milliseconds())
|
||||
return robotRecords, err
|
||||
}
|
||||
@ -156,25 +156,24 @@ func splitRoomRocketProgressDue(entries []roomRocketProgressEntry) ([]roomRocket
|
||||
robotRecords = append(robotRecords, record)
|
||||
continue
|
||||
}
|
||||
entry.record = record
|
||||
normal = append(normal, entry)
|
||||
}
|
||||
return normal, robotRecords
|
||||
}
|
||||
|
||||
func (c *roomRocketProgressCoalescer) saveNormalDue(ctx context.Context, entries []roomRocketProgressEntry) error {
|
||||
normal := make(map[string][]outbox.Record)
|
||||
func (c *roomRocketProgressCoalescer) publishDue(ctx context.Context, entries []roomRocketProgressEntry) error {
|
||||
if c.publisher == nil {
|
||||
return nil
|
||||
}
|
||||
for _, entry := range entries {
|
||||
record := entry.record
|
||||
record.AppCode = appcode.Normalize(record.AppCode)
|
||||
if record.Envelope != nil {
|
||||
record.Envelope.AppCode = record.AppCode
|
||||
if record.Envelope == nil {
|
||||
continue
|
||||
}
|
||||
normal[record.AppCode] = append(normal[record.AppCode], record)
|
||||
}
|
||||
|
||||
for scopedApp, records := range normal {
|
||||
// repository 会按 context app_code 二次归一，分 app 批量写避免跨租户记录被同一个上下文覆盖。
|
||||
if err := c.repository.SaveOutbox(appcode.WithContext(ctx, scopedApp), records); err != nil {
|
||||
record.AppCode = appcode.Normalize(record.AppCode)
|
||||
record.Envelope.AppCode = record.AppCode
|
||||
if err := c.publisher.PublishOutboxEvent(appcode.WithContext(ctx, record.AppCode), record.Envelope); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
@ -199,7 +198,7 @@ func roomRocketProgressWindowStartMS(createdAtMS int64) int64 {
|
||||
return createdAtMS / windowMS * windowMS
|
||||
}
|
||||
|
||||
// FlushRoomRocketProgress 把已经超过 1 秒窗口的火箭进度写回主 outbox；机器人进度只 best-effort 直发 IM。
|
||||
// FlushRoomRocketProgress 把已经超过 1 秒窗口的火箭进度直接发腾讯 IM；机器人 lane 保持原展示直发路径。
|
||||
func (s *Service) FlushRoomRocketProgress(ctx context.Context) error {
|
||||
if s == nil || s.roomRocketProgressCoalescer == nil {
|
||||
return nil
|
||||
@ -209,7 +208,7 @@ func (s *Service) FlushRoomRocketProgress(ctx context.Context) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// RunRoomRocketProgressFlushWorker 周期性把内存合并后的火箭进度落入主 outbox；机器人展示进度不经过 MySQL。
|
||||
// RunRoomRocketProgressFlushWorker 周期性把内存合并后的普通火箭进度直接投递到房间 IM。
|
||||
func (s *Service) RunRoomRocketProgressFlushWorker(ctx context.Context, interval time.Duration) {
|
||||
if interval <= 0 {
|
||||
interval = roomRocketProgressFlushInterval
|
||||
|
||||
@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@ -36,6 +37,37 @@ type rocketTestWallet struct {
|
||||
lastBatch *walletv1.BatchDebitGiftRequest
|
||||
}
|
||||
|
||||
type recordingRoomDirectIMPublisher struct {
|
||||
mu sync.Mutex
|
||||
wake chan struct{}
|
||||
records []*roomeventsv1.EventEnvelope
|
||||
}
|
||||
|
||||
func newRecordingRoomDirectIMPublisher() *recordingRoomDirectIMPublisher {
|
||||
return &recordingRoomDirectIMPublisher{wake: make(chan struct{}, 128)}
|
||||
}
|
||||
|
||||
func (p *recordingRoomDirectIMPublisher) PublishOutboxEvent(_ context.Context, envelope *roomeventsv1.EventEnvelope) error {
|
||||
p.mu.Lock()
|
||||
p.records = append(p.records, proto.Clone(envelope).(*roomeventsv1.EventEnvelope))
|
||||
p.mu.Unlock()
|
||||
select {
|
||||
case p.wake <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *recordingRoomDirectIMPublisher) counts() map[string]int {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
counts := make(map[string]int, len(p.records))
|
||||
for _, record := range p.records {
|
||||
counts[record.GetEventType()]++
|
||||
}
|
||||
return counts
|
||||
}
|
||||
|
||||
func (w *rocketTestWallet) DebitGift(_ context.Context, req *walletv1.DebitGiftRequest) (*walletv1.DebitGiftResponse, error) {
|
||||
w.lastDebit = req
|
||||
w.debitRequests = append(w.debitRequests, req)
|
||||
@ -108,7 +140,8 @@ func TestSendGiftBatchDebitsMultipleTargets(t *testing.T) {
|
||||
{BillingReceiptId: "receipt-202", CoinSpent: 10, ChargeAmount: 10, GiftPointAdded: 5, HeatValue: 11, GiftTypeCode: "normal", PriceVersion: "v1", BalanceAfter: 90},
|
||||
{BillingReceiptId: "receipt-303", CoinSpent: 20, ChargeAmount: 20, GiftPointAdded: 7, HeatValue: 19, GiftTypeCode: "normal", PriceVersion: "v1", BalanceAfter: 70},
|
||||
}}
|
||||
svc := newRocketTestService(t, repository, wallet, now)
|
||||
directIM := newRecordingRoomDirectIMPublisher()
|
||||
svc := newRocketTestServiceWithDirectIM(t, repository, wallet, now, nil, directIM)
|
||||
|
||||
roomID := "room-gift-multi-target"
|
||||
ownerID := int64(101)
|
||||
@ -142,6 +175,9 @@ func TestSendGiftBatchDebitsMultipleTargets(t *testing.T) {
|
||||
if resp.GetRoomHeat() != 30 || resp.GetBillingReceiptId() != "receipt-202,receipt-303" {
|
||||
t.Fatalf("multi-target gift response must aggregate wallet results: %+v", resp)
|
||||
}
|
||||
if resp.GetCoinBalanceAfter() != 70 {
|
||||
t.Fatalf("multi-target gift must return latest sender balance: %d", resp.GetCoinBalanceAfter())
|
||||
}
|
||||
if firstUser := onlineUserByID(resp.GetRoom(), firstTargetID); firstUser == nil || firstUser.GetGiftValue() != 11 {
|
||||
t.Fatalf("first target must receive only its own heat value: %+v", firstUser)
|
||||
}
|
||||
@ -184,6 +220,59 @@ func TestSendGiftBatchDebitsMultipleTargets(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSendGiftMirrorsGiftDisplayToDirectIMAndKeepsDurableFact(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repository := mysqltest.NewRepository(t)
|
||||
now := &fixedRoomRocketClock{now: time.Date(2026, 6, 26, 9, 0, 0, 0, time.UTC)}
|
||||
wallet := &rocketTestWallet{debits: []*walletv1.DebitGiftResponse{{
|
||||
BillingReceiptId: "receipt-direct-im-gift",
|
||||
CoinSpent: 25,
|
||||
ChargeAmount: 25,
|
||||
HeatValue: 25,
|
||||
GiftTypeCode: "normal",
|
||||
BalanceAfter: 975,
|
||||
GiftIncomeBalanceAfter: 120,
|
||||
}}}
|
||||
directIM := newRecordingRoomDirectIMPublisher()
|
||||
svc := newRocketTestServiceWithDirectIM(t, repository, wallet, now, nil, directIM)
|
||||
|
||||
roomID := "room-gift-direct-im"
|
||||
ownerID := int64(701)
|
||||
targetID := int64(702)
|
||||
createRocketRoom(t, ctx, svc, roomID, ownerID, 9001)
|
||||
joinRocketRoom(t, ctx, svc, roomID, targetID)
|
||||
before := outboxEventCounts(t, ctx, repository)
|
||||
|
||||
resp, err := svc.SendGift(ctx, &roomv1.SendGiftRequest{
|
||||
Meta: rocketMeta(roomID, ownerID, "gift-direct-im"),
|
||||
TargetType: "user",
|
||||
TargetUserId: targetID,
|
||||
GiftId: "rose",
|
||||
GiftCount: 1,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("send direct IM gift failed: %v", err)
|
||||
}
|
||||
if resp.GetCoinBalanceAfter() != 975 || resp.GetGiftIncomeBalanceAfter() != 120 {
|
||||
t.Fatalf("gift response must carry balance snapshots: %+v", resp)
|
||||
}
|
||||
|
||||
after := outboxEventCounts(t, ctx, repository)
|
||||
if delta := after["RoomGiftSent"] - before["RoomGiftSent"]; delta != 1 {
|
||||
t.Fatalf("RoomGiftSent durable fact must stay in room_outbox, delta=%d counts=%+v", delta, after)
|
||||
}
|
||||
for _, eventType := range []string{"RoomHeatChanged", "RoomRankChanged"} {
|
||||
if delta := after[eventType] - before[eventType]; delta != 0 {
|
||||
t.Fatalf("%s must not write room_outbox after directIM split, delta=%d counts=%+v", eventType, delta, after)
|
||||
}
|
||||
}
|
||||
waitForRoomDirectIMCounts(t, directIM, map[string]int{
|
||||
"RoomGiftSent": 1,
|
||||
"RoomHeatChanged": 1,
|
||||
"RoomRankChanged": 1,
|
||||
})
|
||||
}
|
||||
|
||||
func (w *rocketTestWallet) GrantResourceGroup(_ context.Context, req *walletv1.GrantResourceGroupRequest) (*walletv1.ResourceGrantResponse, error) {
|
||||
w.grants = append(w.grants, req)
|
||||
return &walletv1.ResourceGrantResponse{
|
||||
@ -212,7 +301,8 @@ func TestRoomRocketGiftOverflowAndImmediateNextLevel(t *testing.T) {
|
||||
{BillingReceiptId: "receipt-fill", GiftPointAdded: 250, HeatValue: 250, GiftTypeCode: "normal"},
|
||||
{BillingReceiptId: "receipt-next", GiftPointAdded: 50, HeatValue: 50, GiftTypeCode: "normal"},
|
||||
}}
|
||||
svc := newRocketTestService(t, repository, wallet, now)
|
||||
directIM := newRecordingRoomDirectIMPublisher()
|
||||
svc := newRocketTestServiceWithDirectIM(t, repository, wallet, now, nil, directIM)
|
||||
writeRocketConfig(t, ctx, repository, rocketConfigForTest(100, 1_000))
|
||||
|
||||
roomID := "room-rocket-flow"
|
||||
@ -250,15 +340,16 @@ func TestRoomRocketGiftOverflowAndImmediateNextLevel(t *testing.T) {
|
||||
}
|
||||
|
||||
flushRocketProgress(t, ctx, svc, now)
|
||||
progressEvents := rocketProgressEvents(t, ctx, repository)
|
||||
progressEvents := roomDirectIMRocketProgressEvents(t, directIM)
|
||||
if len(progressEvents) != 1 || progressEvents[0].GetAddedFuel() != 250 || progressEvents[0].GetEffectiveAddedFuel() != 100 || progressEvents[0].GetOverflowFuel() != 150 {
|
||||
t.Fatalf("progress event must expose effective fuel and discarded overflow: %+v", progressEvents)
|
||||
}
|
||||
if progressEvents[0].GetLevel() != 1 || progressEvents[0].GetStatus() != state.RocketStatusIgnited {
|
||||
t.Fatalf("progress event must describe the filled level, got %+v", progressEvents[0])
|
||||
}
|
||||
if got := countRocketIgnitedEvents(t, ctx, repository); got != 1 {
|
||||
t.Fatalf("fill gift must emit one ignited event, got %d", got)
|
||||
waitForRoomDirectIMCounts(t, directIM, map[string]int{"RoomRocketIgnited": 1})
|
||||
if got := outboxEventCounts(t, ctx, repository)["RoomRocketIgnited"]; got != 1 {
|
||||
t.Fatalf("fill gift must keep one RoomRocketIgnited outbox for launch scheduling, got %d", got)
|
||||
}
|
||||
|
||||
nextResp, err := svc.SendGift(ctx, &roomv1.SendGiftRequest{
|
||||
@ -278,11 +369,11 @@ func TestRoomRocketGiftOverflowAndImmediateNextLevel(t *testing.T) {
|
||||
t.Fatalf("new progress must keep the previous pending launch only: %+v", nextResp.GetRocket())
|
||||
}
|
||||
flushRocketProgress(t, ctx, svc, now)
|
||||
progressEvents = rocketProgressEvents(t, ctx, repository)
|
||||
progressEvents = roomDirectIMRocketProgressEvents(t, directIM)
|
||||
if len(progressEvents) != 2 || progressEvents[1].GetLevel() != 2 || progressEvents[1].GetEffectiveAddedFuel() != 50 || progressEvents[1].GetOverflowFuel() != 0 {
|
||||
t.Fatalf("second progress event must target level 2: %+v", progressEvents)
|
||||
}
|
||||
if got := countRocketIgnitedEvents(t, ctx, repository); got != 1 {
|
||||
if got := directIM.counts()["RoomRocketIgnited"]; got != 1 {
|
||||
t.Fatalf("next-level partial gift must not emit another ignited event, got %d", got)
|
||||
}
|
||||
}
|
||||
@ -296,7 +387,8 @@ func TestRoomRocketProgressCoalescesOneSecondWindow(t *testing.T) {
|
||||
{BillingReceiptId: "receipt-progress-20", GiftPointAdded: 20, HeatValue: 20, GiftTypeCode: "normal"},
|
||||
{BillingReceiptId: "receipt-progress-30", GiftPointAdded: 30, HeatValue: 30, GiftTypeCode: "normal"},
|
||||
}}
|
||||
svc := newRocketTestService(t, repository, wallet, now)
|
||||
directIM := newRecordingRoomDirectIMPublisher()
|
||||
svc := newRocketTestServiceWithDirectIM(t, repository, wallet, now, nil, directIM)
|
||||
writeRocketConfig(t, ctx, repository, rocketConfigForTest(1000, 1_000))
|
||||
|
||||
roomID := "room-rocket-progress-coalesce"
|
||||
@ -319,9 +411,12 @@ func TestRoomRocketProgressCoalescesOneSecondWindow(t *testing.T) {
|
||||
}
|
||||
|
||||
beforeFlush := outboxEventCounts(t, ctx, repository)
|
||||
for _, eventType := range []string{"RoomGiftSent", "RoomHeatChanged", "RoomRankChanged"} {
|
||||
if delta := beforeFlush[eventType] - before[eventType]; delta != 3 {
|
||||
t.Fatalf("%s must stay per gift before progress flush, delta=%d counts=%+v", eventType, delta, beforeFlush)
|
||||
if delta := beforeFlush["RoomGiftSent"] - before["RoomGiftSent"]; delta != 3 {
|
||||
t.Fatalf("RoomGiftSent must stay durable per gift, delta=%d counts=%+v", delta, beforeFlush)
|
||||
}
|
||||
for _, eventType := range []string{"RoomHeatChanged", "RoomRankChanged"} {
|
||||
if delta := beforeFlush[eventType] - before[eventType]; delta != 0 {
|
||||
t.Fatalf("%s must move to direct IM instead of durable outbox, delta=%d counts=%+v", eventType, delta, beforeFlush)
|
||||
}
|
||||
}
|
||||
if delta := beforeFlush["RoomRocketFuelChanged"] - before["RoomRocketFuelChanged"]; delta != 0 {
|
||||
@ -330,13 +425,10 @@ func TestRoomRocketProgressCoalescesOneSecondWindow(t *testing.T) {
|
||||
|
||||
flushRocketProgress(t, ctx, svc, now)
|
||||
afterFlush := outboxEventCounts(t, ctx, repository)
|
||||
if delta := afterFlush["RoomRocketFuelChanged"] - before["RoomRocketFuelChanged"]; delta != 1 {
|
||||
t.Fatalf("rocket progress must coalesce to one outbox row, delta=%d counts=%+v", delta, afterFlush)
|
||||
}
|
||||
progressEvents := rocketProgressEvents(t, ctx, repository)
|
||||
if len(progressEvents) != 1 || progressEvents[0].GetCurrentFuel() != 60 || progressEvents[0].GetEffectiveAddedFuel() != 30 {
|
||||
t.Fatalf("coalesced progress must keep latest max fuel event: %+v", progressEvents)
|
||||
if delta := afterFlush["RoomRocketFuelChanged"] - before["RoomRocketFuelChanged"]; delta != 0 {
|
||||
t.Fatalf("rocket progress must flush to direct IM instead of durable outbox, delta=%d counts=%+v", delta, afterFlush)
|
||||
}
|
||||
waitForRoomDirectIMCounts(t, directIM, map[string]int{"RoomRocketFuelChanged": 1})
|
||||
}
|
||||
|
||||
func TestRoomRocketIgnitedOutboxSchedulesDelayedLaunch(t *testing.T) {
|
||||
@ -788,6 +880,11 @@ func newRocketTestService(t *testing.T, repository *mysqltest.Repository, wallet
|
||||
|
||||
func newRocketTestServiceWithScheduler(t *testing.T, repository *mysqltest.Repository, wallet *rocketTestWallet, clock *fixedRoomRocketClock, scheduler integration.RoomRocketLaunchScheduler) *roomservice.Service {
|
||||
t.Helper()
|
||||
return newRocketTestServiceWithDirectIM(t, repository, wallet, clock, scheduler, nil)
|
||||
}
|
||||
|
||||
func newRocketTestServiceWithDirectIM(t *testing.T, repository *mysqltest.Repository, wallet *rocketTestWallet, clock *fixedRoomRocketClock, scheduler integration.RoomRocketLaunchScheduler, directIM integration.OutboxPublisher) *roomservice.Service {
|
||||
t.Helper()
|
||||
|
||||
return roomservice.New(roomservice.Config{
|
||||
NodeID: "node-rocket-test",
|
||||
@ -796,9 +893,34 @@ func newRocketTestServiceWithScheduler(t *testing.T, repository *mysqltest.Repos
|
||||
SnapshotEveryN: 1,
|
||||
Clock: clock,
|
||||
RoomRocketLaunchScheduler: scheduler,
|
||||
RoomDisplayPublisher: directIM,
|
||||
}, router.NewMemoryDirectory(), repository, wallet, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher())
|
||||
}
|
||||
|
||||
func waitForRoomDirectIMCounts(t testing.TB, publisher *recordingRoomDirectIMPublisher, want map[string]int) {
|
||||
t.Helper()
|
||||
deadline := time.NewTimer(2 * time.Second)
|
||||
defer deadline.Stop()
|
||||
for {
|
||||
counts := publisher.counts()
|
||||
matched := true
|
||||
for eventType, expected := range want {
|
||||
if counts[eventType] < expected {
|
||||
matched = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if matched {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-publisher.wake:
|
||||
case <-deadline.C:
|
||||
t.Fatalf("room direct IM events mismatch: want at least %+v got %+v", want, counts)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func flushRocketProgress(t testing.TB, ctx context.Context, svc *roomservice.Service, clock *fixedRoomRocketClock) {
|
||||
t.Helper()
|
||||
|
||||
@ -921,6 +1043,25 @@ func rocketProgressEvents(t *testing.T, ctx context.Context, repository *mysqlte
|
||||
return events
|
||||
}
|
||||
|
||||
func roomDirectIMRocketProgressEvents(t *testing.T, publisher *recordingRoomDirectIMPublisher) []*roomeventsv1.RoomRocketFuelChanged {
|
||||
t.Helper()
|
||||
publisher.mu.Lock()
|
||||
records := append([]*roomeventsv1.EventEnvelope(nil), publisher.records...)
|
||||
publisher.mu.Unlock()
|
||||
events := make([]*roomeventsv1.RoomRocketFuelChanged, 0)
|
||||
for _, record := range records {
|
||||
if record.GetEventType() != "RoomRocketFuelChanged" {
|
||||
continue
|
||||
}
|
||||
var event roomeventsv1.RoomRocketFuelChanged
|
||||
if err := proto.Unmarshal(record.GetBody(), &event); err != nil {
|
||||
t.Fatalf("decode directIM rocket progress failed: %v", err)
|
||||
}
|
||||
events = append(events, &event)
|
||||
}
|
||||
return events
|
||||
}
|
||||
|
||||
func roomGiftSentEvents(t *testing.T, ctx context.Context, repository *mysqltest.Repository) []*roomeventsv1.RoomGiftSent {
|
||||
t.Helper()
|
||||
|
||||
|
||||
@ -37,6 +37,8 @@ type Config struct {
|
||||
RoomRocketLaunchScheduler integration.RoomRocketLaunchScheduler
|
||||
// RobotDisplayPublisher 专门 best-effort 投递机器人房间展示事件；失败只打日志，不进入 MySQL outbox。
|
||||
RobotDisplayPublisher integration.OutboxPublisher
|
||||
// RoomDisplayPublisher 在 Room Cell 提交后直接投递腾讯 IM 展示事件；失败不回滚已提交房间命令。
|
||||
RoomDisplayPublisher integration.OutboxPublisher
|
||||
// HumanRobotPoolProvider 自动读取全站机器人账号和用户国家，用于真人房间机器人运行时按国家分组。
|
||||
HumanRobotPoolProvider integration.HumanRoomRobotPoolProvider
|
||||
// RoomGiftLeaderboard 是跨房间贡献榜读模型，SendGift 提交后 best-effort 写入。
|
||||
@ -81,6 +83,8 @@ type Service struct {
|
||||
outboxPublisher integration.OutboxPublisher
|
||||
// robotDisplayPublisher 只服务机器人房间展示事件，和主 room_outbox 持久事实隔离。
|
||||
robotDisplayPublisher integration.OutboxPublisher
|
||||
// roomDisplayPublisher 只承载客户端可由快照修复的直接 IM 展示事件，不作为业务事实源。
|
||||
roomDisplayPublisher integration.OutboxPublisher
|
||||
// outboxWake 提交成功后唤醒主 outbox worker；它只是低延迟提示，事实源仍然是 MySQL room_outbox。
|
||||
outboxWake chan struct{}
|
||||
// humanRobotPoolProvider 运行时自动获取全站机器人池，避免后台手工维护国家机器人池。
|
||||
@ -150,6 +154,10 @@ type mutationResult struct {
|
||||
micHeartbeatAtMS int64
|
||||
// billingReceiptID 是 SendGift 从 wallet-service 得到的账务回执。
|
||||
billingReceiptID string
|
||||
// coinBalanceAfter 是送礼人可展示的最终 COIN 余额；幸运礼物返奖成功时优先使用返奖后的余额。
|
||||
coinBalanceAfter int64
|
||||
// giftIncomeBalanceAfter 是收礼人返币后的 COIN 余额；单目标返币可同步返回给发起方确认。
|
||||
giftIncomeBalanceAfter int64
|
||||
// roomHeat 是 SendGift 后的房间热度。
|
||||
roomHeat int64
|
||||
// giftRank 是 SendGift 后的本地礼物榜投影。
|
||||
@ -160,12 +168,14 @@ type mutationResult struct {
|
||||
luckyGifts []*roomv1.LuckyGiftDrawResult
|
||||
// commandPayload 允许少数命令把外部依赖的结算快照写入 command log，用于恢复。
|
||||
commandPayload []byte
|
||||
// syncEvent 标记该命令存在腾讯云 IM 房间消息形态；实际投递只由 outbox worker 异步执行。
|
||||
// syncEvent 标记该命令存在腾讯云 IM 房间消息形态；保留旧字段兼容命令结果，room UI 现在由提交后 direct IM 发送。
|
||||
syncEvent *tencentim.RoomEvent
|
||||
// outboxRecords 是本次命令已经持久化的房间事实，提交后可投给 activity 等外部消费者。
|
||||
outboxRecords []outbox.Record
|
||||
// robotDisplayRecords 是机器人房间展示事实，提交成功后 best-effort 直发 Tencent IM，不占主 room_outbox。
|
||||
robotDisplayRecords []outbox.Record
|
||||
// directIMRecords 是提交成功后直接发腾讯 IM 的展示事件；失败不回滚已提交房间命令。
|
||||
directIMRecords []outbox.Record
|
||||
// roomRocketProgress 是提交成功后才进入内存合并器的火箭进度展示事件，不和送礼命令事务逐笔写 outbox。
|
||||
roomRocketProgress *roomRocketProgressPending
|
||||
// roomStatus 非空时代表命令提交时需要同步更新 rooms 和列表投影生命周期状态。
|
||||
@ -243,10 +253,11 @@ func New(cfg Config, directory router.Directory, repository Repository, wallet i
|
||||
syncPublisher: syncPublisher,
|
||||
outboxPublisher: outboxPublisher,
|
||||
robotDisplayPublisher: cfg.RobotDisplayPublisher,
|
||||
roomDisplayPublisher: cfg.RoomDisplayPublisher,
|
||||
outboxWake: make(chan struct{}, 1024),
|
||||
humanRobotPoolProvider: cfg.HumanRobotPoolProvider,
|
||||
roomRocketLaunchScheduler: cfg.RoomRocketLaunchScheduler,
|
||||
roomRocketProgressCoalescer: newRoomRocketProgressCoalescer(repository),
|
||||
roomRocketProgressCoalescer: newRoomRocketProgressCoalescer(cfg.RoomDisplayPublisher),
|
||||
robotDisplaySampler: newRobotDisplaySampler(),
|
||||
roomGiftLeaderboard: cfg.RoomGiftLeaderboard,
|
||||
roomListCache: cfg.RoomListCache,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user