From afc62eca1e0305902929c0f114a6a5f46f24695d Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 1 Jun 2026 16:09:48 +0800 Subject: [PATCH] fix: credit lucky gift rewards synchronously --- api/proto/activity/v1/activity.pb.go | 22 ++- api/proto/activity/v1/activity.proto | 2 + api/proto/room/v1/room.pb.go | 22 ++- api/proto/room/v1/room.proto | 2 + .../internal/domain/luckygift/lucky_gift.go | 8 + .../internal/service/luckygift/service.go | 133 ++++++++++++--- .../service/luckygift/service_test.go | 152 ++++++++++++++++- .../storage/mysql/lucky_gift_repository.go | 64 ++++++- .../transport/grpc/lucky_gift_server.go | 2 + .../internal/transport/http/response_test.go | 9 +- .../room-service/configs/config.docker.yaml | 1 + .../configs/config.tencent.example.yaml | 1 + services/room-service/configs/config.yaml | 1 + services/room-service/internal/app/app.go | 2 + .../room-service/internal/config/config.go | 32 ++-- .../internal/room/service/gift.go | 158 +++++++++++------- .../room/service/lucky_gift_send_lock.go | 60 +++++++ .../room/service/lucky_gift_send_lock_test.go | 59 +++++++ .../internal/room/service/lucky_gift_test.go | 6 +- .../internal/room/service/service.go | 14 ++ 20 files changed, 636 insertions(+), 114 deletions(-) create mode 100644 services/room-service/internal/room/service/lucky_gift_send_lock.go create mode 100644 services/room-service/internal/room/service/lucky_gift_send_lock_test.go diff --git a/api/proto/activity/v1/activity.pb.go b/api/proto/activity/v1/activity.pb.go index 01eaa0bd..38c67b6a 100644 --- a/api/proto/activity/v1/activity.pb.go +++ b/api/proto/activity/v1/activity.pb.go @@ -11087,6 +11087,8 @@ type LuckyGiftDrawResult struct { CreatedAtMs int64 `protobuf:"varint,19,opt,name=created_at_ms,json=createdAtMs,proto3" json:"created_at_ms,omitempty"` PoolId string `protobuf:"bytes,20,opt,name=pool_id,json=poolId,proto3" json:"pool_id,omitempty"` MultiplierPpm int64 `protobuf:"varint,21,opt,name=multiplier_ppm,json=multiplierPpm,proto3" json:"multiplier_ppm,omitempty"` + WalletTransactionId string `protobuf:"bytes,22,opt,name=wallet_transaction_id,json=walletTransactionId,proto3" json:"wallet_transaction_id,omitempty"` + CoinBalanceAfter int64 `protobuf:"varint,23,opt,name=coin_balance_after,json=coinBalanceAfter,proto3" json:"coin_balance_after,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -11268,6 +11270,20 @@ func (x *LuckyGiftDrawResult) GetMultiplierPpm() int64 { return 0 } +func (x *LuckyGiftDrawResult) GetWalletTransactionId() string { + if x != nil { + return x.WalletTransactionId + } + return "" +} + +func (x *LuckyGiftDrawResult) GetCoinBalanceAfter() int64 { + if x != nil { + return x.CoinBalanceAfter + } + return 0 +} + type ExecuteLuckyGiftDrawRequest struct { state protoimpl.MessageState `protogen:"open.v1"` LuckyGift *LuckyGiftMeta `protobuf:"bytes,1,opt,name=lucky_gift,json=luckyGift,proto3" json:"lucky_gift,omitempty"` @@ -13101,7 +13117,7 @@ const file_proto_activity_v1_activity_proto_rawDesc = "" + "\frule_version\x18\x05 \x01(\x03R\vruleVersion\x12$\n" + "\x0etarget_rtp_ppm\x18\x06 \x01(\x03R\ftargetRtpPpm\x12'\n" + "\x0fexperience_pool\x18\a \x01(\tR\x0eexperiencePool\x12\x17\n" + - "\apool_id\x18\b \x01(\tR\x06poolId\"\xf5\x06\n" + + "\apool_id\x18\b \x01(\tR\x06poolId\"\xd7\a\n" + "\x13LuckyGiftDrawResult\x12\x17\n" + "\adraw_id\x18\x01 \x01(\tR\x06drawId\x12\x1d\n" + "\n" + @@ -13125,7 +13141,9 @@ const file_proto_activity_v1_activity_proto_rawDesc = "" + "\x0fhigh_multiplier\x18\x12 \x01(\bR\x0ehighMultiplier\x12\"\n" + "\rcreated_at_ms\x18\x13 \x01(\x03R\vcreatedAtMs\x12\x17\n" + "\apool_id\x18\x14 \x01(\tR\x06poolId\x12%\n" + - "\x0emultiplier_ppm\x18\x15 \x01(\x03R\rmultiplierPpm\"^\n" + + "\x0emultiplier_ppm\x18\x15 \x01(\x03R\rmultiplierPpm\x122\n" + + "\x15wallet_transaction_id\x18\x16 \x01(\tR\x13walletTransactionId\x12,\n" + + "\x12coin_balance_after\x18\x17 \x01(\x03R\x10coinBalanceAfter\"^\n" + "\x1bExecuteLuckyGiftDrawRequest\x12?\n" + "\n" + "lucky_gift\x18\x01 \x01(\v2 .hyapp.activity.v1.LuckyGiftMetaR\tluckyGift\"^\n" + diff --git a/api/proto/activity/v1/activity.proto b/api/proto/activity/v1/activity.proto index 3d64d44f..57f07001 100644 --- a/api/proto/activity/v1/activity.proto +++ b/api/proto/activity/v1/activity.proto @@ -1263,6 +1263,8 @@ message LuckyGiftDrawResult { int64 created_at_ms = 19; string pool_id = 20; int64 multiplier_ppm = 21; + string wallet_transaction_id = 22; + int64 coin_balance_after = 23; } message ExecuteLuckyGiftDrawRequest { diff --git a/api/proto/room/v1/room.pb.go b/api/proto/room/v1/room.pb.go index 4e8bb3c2..a3f95753 100644 --- a/api/proto/room/v1/room.pb.go +++ b/api/proto/room/v1/room.pb.go @@ -562,6 +562,8 @@ type LuckyGiftDrawResult struct { StageFeedback bool `protobuf:"varint,15,opt,name=stage_feedback,json=stageFeedback,proto3" json:"stage_feedback,omitempty"` HighMultiplier bool `protobuf:"varint,16,opt,name=high_multiplier,json=highMultiplier,proto3" json:"high_multiplier,omitempty"` CreatedAtMs int64 `protobuf:"varint,17,opt,name=created_at_ms,json=createdAtMs,proto3" json:"created_at_ms,omitempty"` + WalletTransactionId string `protobuf:"bytes,18,opt,name=wallet_transaction_id,json=walletTransactionId,proto3" json:"wallet_transaction_id,omitempty"` + CoinBalanceAfter int64 `protobuf:"varint,19,opt,name=coin_balance_after,json=coinBalanceAfter,proto3" json:"coin_balance_after,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -715,6 +717,20 @@ func (x *LuckyGiftDrawResult) GetCreatedAtMs() int64 { return 0 } +func (x *LuckyGiftDrawResult) GetWalletTransactionId() string { + if x != nil { + return x.WalletTransactionId + } + return "" +} + +func (x *LuckyGiftDrawResult) GetCoinBalanceAfter() int64 { + if x != nil { + return x.CoinBalanceAfter + } + return 0 +} + // RoomTreasureRewardItem 是后台配置给客户端展示的宝箱奖励候选项。 type RoomTreasureRewardItem struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -8881,7 +8897,7 @@ const file_proto_room_v1_room_proto_rawDesc = "" + "\x05score\x18\x02 \x01(\x03R\x05score\x12\x1d\n" + "\n" + "gift_value\x18\x03 \x01(\x03R\tgiftValue\x12\"\n" + - "\rupdated_at_ms\x18\x04 \x01(\x03R\vupdatedAtMs\"\xa8\x05\n" + + "\rupdated_at_ms\x18\x04 \x01(\x03R\vupdatedAtMs\"\x8a\x06\n" + "\x13LuckyGiftDrawResult\x12\x18\n" + "\aenabled\x18\x01 \x01(\bR\aenabled\x12\x17\n" + "\adraw_id\x18\x02 \x01(\tR\x06drawId\x12\x1d\n" + @@ -8901,7 +8917,9 @@ const file_proto_room_v1_room_proto_rawDesc = "" + "\rreward_status\x18\x0e \x01(\tR\frewardStatus\x12%\n" + "\x0estage_feedback\x18\x0f \x01(\bR\rstageFeedback\x12'\n" + "\x0fhigh_multiplier\x18\x10 \x01(\bR\x0ehighMultiplier\x12\"\n" + - "\rcreated_at_ms\x18\x11 \x01(\x03R\vcreatedAtMs\"\xc0\x01\n" + + "\rcreated_at_ms\x18\x11 \x01(\x03R\vcreatedAtMs\x122\n" + + "\x15wallet_transaction_id\x18\x12 \x01(\tR\x13walletTransactionId\x12,\n" + + "\x12coin_balance_after\x18\x13 \x01(\x03R\x10coinBalanceAfter\"\xc0\x01\n" + "\x16RoomTreasureRewardItem\x12$\n" + "\x0ereward_item_id\x18\x01 \x01(\tR\frewardItemId\x12*\n" + "\x11resource_group_id\x18\x02 \x01(\x03R\x0fresourceGroupId\x12\x16\n" + diff --git a/api/proto/room/v1/room.proto b/api/proto/room/v1/room.proto index fb6360e3..74872eb0 100644 --- a/api/proto/room/v1/room.proto +++ b/api/proto/room/v1/room.proto @@ -94,6 +94,8 @@ message LuckyGiftDrawResult { bool stage_feedback = 15; bool high_multiplier = 16; int64 created_at_ms = 17; + string wallet_transaction_id = 18; + int64 coin_balance_after = 19; } // RoomTreasureRewardItem 是后台配置给客户端展示的宝箱奖励候选项。 diff --git a/services/activity-service/internal/domain/luckygift/lucky_gift.go b/services/activity-service/internal/domain/luckygift/lucky_gift.go index 1c0a4878..62cc60bd 100644 --- a/services/activity-service/internal/domain/luckygift/lucky_gift.go +++ b/services/activity-service/internal/domain/luckygift/lucky_gift.go @@ -164,6 +164,7 @@ type CheckResult struct { type DrawResult struct { DrawID string + DrawIDs []string CommandID string PoolID string GiftID string @@ -184,6 +185,8 @@ type DrawResult struct { StageFeedback bool HighMultiplier bool CreatedAtMS int64 + WalletTransactionID string + CoinBalanceAfter int64 } type DrawSummary struct { @@ -225,3 +228,8 @@ type DrawOutbox struct { RetryCount int CreatedAtMS int64 } + +type DrawRewardState struct { + AllGranted bool + WalletTransactionID string +} diff --git a/services/activity-service/internal/service/luckygift/service.go b/services/activity-service/internal/service/luckygift/service.go index 4f7ea5e8..8c0eba38 100644 --- a/services/activity-service/internal/service/luckygift/service.go +++ b/services/activity-service/internal/service/luckygift/service.go @@ -30,6 +30,7 @@ type Repository interface { ExecuteLuckyGiftDraw(ctx context.Context, cmd domain.DrawCommand, nowMS int64) (domain.DrawResult, error) ListLuckyGiftDraws(ctx context.Context, query domain.DrawQuery) ([]domain.DrawResult, int64, error) GetLuckyGiftDrawSummary(ctx context.Context, query domain.DrawQuery) (domain.DrawSummary, error) + GetLuckyGiftDrawRewardState(ctx context.Context, appCode string, drawIDs []string) (domain.DrawRewardState, error) ClaimPendingLuckyGiftOutbox(ctx context.Context, workerID string, nowMS int64, lockTTL time.Duration, batchSize int) ([]domain.DrawOutbox, error) MarkLuckyGiftOutboxDelivered(ctx context.Context, event domain.DrawOutbox, nowMS int64) error MarkLuckyGiftOutboxRetryable(ctx context.Context, event domain.DrawOutbox, retryCount int, nextRetryAtMS int64, lastErr string, nowMS int64) error @@ -136,7 +137,11 @@ func (s *Service) Draw(ctx context.Context, cmd domain.DrawCommand) (domain.Draw // paid_at_ms 是扣费事实时间，缺失时用 UTC 服务端时间补齐，避免使用本地时区切风控窗口。 cmd.PaidAtMS = s.now().UTC().UnixMilli() } - return s.repository.ExecuteLuckyGiftDraw(ctx, cmd, s.now().UTC().UnixMilli()) + result, err := s.repository.ExecuteLuckyGiftDraw(ctx, cmd, s.now().UTC().UnixMilli()) + if err != nil { + return domain.DrawResult{}, err + } + return s.creditLuckyGiftRewardFastPath(ctx, cmd, result), nil } type WorkerOptions struct { @@ -235,30 +240,26 @@ func (s *Service) processDrawOutbox(ctx context.Context, event domain.DrawOutbox eventCtx := appcode.WithContext(ctx, event.AppCode) walletTransactionID := "" credited := false + alreadyGranted := false if payload.EffectiveRewardCoins > 0 { - if s.wallet == nil { - return s.markOutboxFailed(eventCtx, event, options, fmt.Errorf("lucky gift wallet client is not configured"), false) - } - // 返奖 command_id 派生自聚合 draw_id，钱包侧可用它做幂等；批量抽奖不按子抽逐条加币，避免待发放和钱包流水膨胀。 - resp, err := s.wallet.CreditLuckyGiftReward(eventCtx, &walletv1.CreditLuckyGiftRewardRequest{ - CommandId: "lucky_reward:" + payload.DrawID, - AppCode: event.AppCode, - TargetUserId: payload.UserID, - Amount: payload.EffectiveRewardCoins, - DrawId: payload.DrawID, - RoomId: payload.RoomID, - GiftId: payload.GiftID, - PoolId: payload.PoolID, - Reason: "lucky_gift_reward", - VisibleRegionId: payload.VisibleRegionID, - }) + state, err := s.repository.GetLuckyGiftDrawRewardState(eventCtx, event.AppCode, payload.DrawIDs) if err != nil { return s.markOutboxFailed(eventCtx, event, options, err, false) } - walletTransactionID = resp.GetTransactionId() - credited = true + alreadyGranted = state.AllGranted + if alreadyGranted { + walletTransactionID = state.WalletTransactionID + credited = true + } else { + receipt, err := s.creditLuckyGiftReward(eventCtx, event.AppCode, payload) + if err != nil { + return s.markOutboxFailed(eventCtx, event, options, err, false) + } + walletTransactionID = receipt.WalletTransactionID + credited = true + } } - if credited { + if credited && !alreadyGranted { nowMS := s.now().UTC().UnixMilli() if err := s.repository.MarkLuckyGiftDrawsGranted(eventCtx, event.AppCode, payload.DrawIDs, walletTransactionID, nowMS); err != nil { return err @@ -289,6 +290,88 @@ func (s *Service) processDrawOutbox(ctx context.Context, event domain.DrawOutbox return s.repository.MarkLuckyGiftOutboxDelivered(eventCtx, event, nowMS) } +type luckyGiftRewardReceipt struct { + WalletTransactionID string + CoinBalanceAfter int64 +} + +func (s *Service) creditLuckyGiftRewardFastPath(ctx context.Context, cmd domain.DrawCommand, result domain.DrawResult) domain.DrawResult { + if result.EffectiveRewardCoins <= 0 || result.RewardStatus == domain.StatusGranted { + return result + } + payload := luckyGiftDrawnPayload{ + AppCode: appcode.FromContext(ctx), + DrawID: result.DrawID, + DrawIDs: luckyDrawResultIDs(result), + CommandID: cmd.CommandID, + PoolID: result.PoolID, + RoomID: cmd.RoomID, + GiftID: cmd.GiftID, + GiftCount: cmd.GiftCount, + UserID: cmd.UserID, + SenderUserID: cmd.UserID, + TargetUserID: cmd.TargetUserID, + VisibleRegionID: cmd.VisibleRegionID, + CoinSpent: cmd.CoinSpent, + RuleVersion: result.RuleVersion, + ExperiencePool: result.ExperiencePool, + SelectedTierID: result.SelectedTierID, + MultiplierPPM: result.MultiplierPPM, + BaseRewardCoins: result.BaseRewardCoins, + RoomAtmosphereRewardCoins: result.RoomAtmosphereRewardCoins, + ActivitySubsidyCoins: result.ActivitySubsidyCoins, + EffectiveRewardCoins: result.EffectiveRewardCoins, + StageFeedback: result.StageFeedback, + HighMultiplier: result.HighMultiplier, + CreatedAtMS: result.CreatedAtMS, + } + receipt, err := s.creditLuckyGiftReward(ctx, payload.AppCode, payload) + if err != nil { + logx.Error(ctx, "lucky_gift_reward_fast_path_failed", err, + slog.String("draw_id", result.DrawID), + slog.String("command_id", result.CommandID), + slog.Int64("reward_coins", result.EffectiveRewardCoins), + ) + return result + } + result.RewardStatus = domain.StatusGranted + result.WalletTransactionID = receipt.WalletTransactionID + result.CoinBalanceAfter = receipt.CoinBalanceAfter + if err := s.repository.MarkLuckyGiftDrawsGranted(ctx, payload.AppCode, payload.DrawIDs, receipt.WalletTransactionID, s.now().UTC().UnixMilli()); err != nil { + logx.Error(ctx, "lucky_gift_reward_fast_path_mark_granted_failed", err, + slog.String("draw_id", result.DrawID), + slog.String("command_id", result.CommandID), + slog.String("wallet_transaction_id", receipt.WalletTransactionID), + ) + } + return result +} + +func (s *Service) creditLuckyGiftReward(ctx context.Context, appCode string, payload luckyGiftDrawnPayload) (luckyGiftRewardReceipt, error) { + if s.wallet == nil { + return luckyGiftRewardReceipt{}, fmt.Errorf("lucky gift wallet client is not configured") + } + resp, err := s.wallet.CreditLuckyGiftReward(appcode.WithContext(ctx, appCode), &walletv1.CreditLuckyGiftRewardRequest{ + CommandId: "lucky_reward:" + payload.DrawID, + AppCode: appCode, + TargetUserId: payload.UserID, + Amount: payload.EffectiveRewardCoins, + DrawId: payload.DrawID, + RoomId: payload.RoomID, + GiftId: payload.GiftID, + PoolId: payload.PoolID, + Reason: "lucky_gift_reward", + VisibleRegionId: payload.VisibleRegionID, + }) + if err != nil { + return luckyGiftRewardReceipt{}, err + } + return luckyGiftRewardReceipt{ + WalletTransactionID: resp.GetTransactionId(), + CoinBalanceAfter: resp.GetBalance().GetAvailableAmount(), + }, nil +} + func (s *Service) publishLuckyGiftDrawn(ctx context.Context, payload luckyGiftDrawnPayload, options WorkerOptions) error { body, err := json.Marshal(luckyGiftRoomMessagePayload(payload)) if err != nil { @@ -554,6 +637,16 @@ func drawIDsFromPayload(payloadJSON string) []string { return payload.DrawIDs } +func luckyDrawResultIDs(result domain.DrawResult) []string { + if len(result.DrawIDs) > 0 { + return append([]string(nil), result.DrawIDs...) + } + if strings.TrimSpace(result.DrawID) == "" { + return nil + } + return []string{result.DrawID} +} + func normalizeWorkerOptions(options WorkerOptions) WorkerOptions { options.WorkerID = strings.TrimSpace(options.WorkerID) if options.WorkerID == "" { diff --git a/services/activity-service/internal/service/luckygift/service_test.go b/services/activity-service/internal/service/luckygift/service_test.go index ee53b26a..8602dd9e 100644 --- a/services/activity-service/internal/service/luckygift/service_test.go +++ b/services/activity-service/internal/service/luckygift/service_test.go @@ -48,6 +48,85 @@ func TestUpsertConfigPreservesPoolScope(t *testing.T) { } } +func TestDrawCreditsLuckyGiftRewardFastPath(t *testing.T) { + repository := &fakeLuckyGiftRepository{executeResult: domain.DrawResult{ + DrawID: "lucky_draw_fast_1", + DrawIDs: []string{"lucky_draw_fast_1", "lucky_draw_fast_2"}, + CommandID: "cmd-gift-fast", + PoolID: "super_lucky", + GiftID: "rose", + RuleVersion: 12, + ExperiencePool: "novice", + SelectedTierID: "batch", + MultiplierPPM: 3_000_000, + EffectiveRewardCoins: 300, + RewardStatus: domain.StatusPending, + CreatedAtMS: 1760000000000, + }} + wallet := &fakeLuckyGiftWallet{balanceAfter: 1300} + service := New(repository, WithWallet(wallet)) + service.SetClock(func() time.Time { return time.UnixMilli(1760000001000).UTC() }) + + result, err := service.Draw(appcode.WithContext(context.Background(), "lalu"), domain.DrawCommand{ + CommandID: "cmd-gift-fast", + PoolID: "super_lucky", + UserID: 42, + TargetUserID: 99, + DeviceID: "device-1", + RoomID: "room-1", + AnchorID: "99", + GiftID: "rose", + GiftCount: 2, + CoinSpent: 200, + PaidAtMS: 1760000000000, + }) + if err != nil { + t.Fatalf("Draw failed: %v", err) + } + if wallet.last == nil || wallet.last.GetCommandId() != "lucky_reward:lucky_draw_fast_1" || wallet.last.GetAmount() != 300 || wallet.last.GetTargetUserId() != 42 { + t.Fatalf("wallet fast path request mismatch: %+v", wallet.last) + } + if result.RewardStatus != domain.StatusGranted || result.WalletTransactionID != "wallet_tx_lucky" || result.CoinBalanceAfter != 1300 { + t.Fatalf("fast path result mismatch: %+v", result) + } + if got := strings.Join(repository.grantedDrawIDs, ","); got != "lucky_draw_fast_1,lucky_draw_fast_2" { + t.Fatalf("fast path should mark all draw ids granted, got %s", got) + } +} + +func TestDrawKeepsPendingWhenFastPathWalletFails(t *testing.T) { + repository := &fakeLuckyGiftRepository{executeResult: domain.DrawResult{ + DrawID: "lucky_draw_retry", + DrawIDs: []string{"lucky_draw_retry"}, + CommandID: "cmd-gift-retry", + PoolID: "super_lucky", + GiftID: "rose", + EffectiveRewardCoins: 300, + RewardStatus: domain.StatusPending, + }} + wallet := &fakeLuckyGiftWallet{err: errors.New("wallet timeout")} + service := New(repository, WithWallet(wallet)) + + result, err := service.Draw(appcode.WithContext(context.Background(), "lalu"), domain.DrawCommand{ + CommandID: "cmd-gift-retry", + PoolID: "super_lucky", + UserID: 42, + DeviceID: "device-1", + RoomID: "room-1", + AnchorID: "99", + GiftID: "rose", + GiftCount: 1, + CoinSpent: 100, + PaidAtMS: 1760000000000, + }) + if err != nil { + t.Fatalf("Draw should keep the committed draw fact: %v", err) + } + if result.RewardStatus != domain.StatusPending || result.WalletTransactionID != "" || len(repository.grantedDrawIDs) != 0 { + t.Fatalf("wallet failure must leave draw pending for outbox compensation: result=%+v granted=%v", result, repository.grantedDrawIDs) + } +} + func TestProcessPendingDrawOutboxCreditsWalletAndPublishesRoomIM(t *testing.T) { payload, _ := json.Marshal(map[string]any{ "event_type": "lucky_gift_drawn", @@ -261,9 +340,60 @@ func TestProcessPendingDrawOutboxGrantsRewardWhenRoomIMFails(t *testing.T) { } } +func TestProcessPendingDrawOutboxSkipsWalletWhenDrawAlreadyGranted(t *testing.T) { + payload, _ := json.Marshal(map[string]any{ + "event_type": "lucky_gift_drawn", + "event_id": "lucky_gift_drawn:lucky_draw_granted", + "app_code": "lalu", + "draw_id": "lucky_draw_granted", + "command_id": "cmd-gift-granted", + "pool_id": "super_lucky", + "user_id": 42, + "room_id": "room-1", + "gift_id": "rose", + "gift_count": 1, + "coin_spent": 100, + "effective_reward_coins": 300, + "multiplier_ppm": 3_000_000, + "created_at_ms": 1760000000000, + }) + repository := &fakeLuckyGiftRepository{ + rewardState: domain.DrawRewardState{AllGranted: true, WalletTransactionID: "wallet_tx_existing"}, + outbox: []domain.DrawOutbox{{ + AppCode: "lalu", + OutboxID: "lucky_lucky_draw_granted", + EventType: "LuckyGiftDrawn", + PayloadJSON: string(payload), + }}, + } + wallet := &fakeLuckyGiftWallet{} + publisher := &fakeLuckyGiftPublisher{} + service := New(repository, WithWallet(wallet), WithRoomPublisher(publisher), WithRegionBroadcaster(&fakeLuckyGiftRegionBroadcaster{})) + + processed, err := service.ProcessPendingDrawOutbox(context.Background(), WorkerOptions{WorkerID: "worker-1"}) + if err != nil { + t.Fatalf("ProcessPendingDrawOutbox failed: %v", err) + } + if processed != 1 { + t.Fatalf("processed count mismatch: %d", processed) + } + if wallet.last != nil { + t.Fatalf("already granted draw must not call wallet again: %+v", wallet.last) + } + if repository.grantedDrawID != "" { + t.Fatalf("already granted draw should not be marked again, got %q", repository.grantedDrawID) + } + if publisher.last.EventID != "lucky_gift_drawn:lucky_draw_granted" || repository.deliveredOutboxID != "lucky_lucky_draw_granted" { + t.Fatalf("already granted draw should still publish IM and deliver outbox: publisher=%+v delivered=%s", publisher.last, repository.deliveredOutboxID) + } +} + type fakeLuckyGiftRepository struct { getConfigCalls int upserted domain.RuleConfig + executeResult domain.DrawResult + executeCmd domain.DrawCommand + rewardState domain.DrawRewardState outbox []domain.DrawOutbox grantedDrawID string @@ -290,8 +420,9 @@ func (r *fakeLuckyGiftRepository) CheckLuckyGift(context.Context, domain.CheckCo return domain.CheckResult{}, nil } -func (r *fakeLuckyGiftRepository) ExecuteLuckyGiftDraw(context.Context, domain.DrawCommand, int64) (domain.DrawResult, error) { - return domain.DrawResult{}, nil +func (r *fakeLuckyGiftRepository) ExecuteLuckyGiftDraw(_ context.Context, cmd domain.DrawCommand, _ int64) (domain.DrawResult, error) { + r.executeCmd = cmd + return r.executeResult, nil } func (r *fakeLuckyGiftRepository) ListLuckyGiftDraws(context.Context, domain.DrawQuery) ([]domain.DrawResult, int64, error) { @@ -302,6 +433,10 @@ func (r *fakeLuckyGiftRepository) GetLuckyGiftDrawSummary(context.Context, domai return domain.DrawSummary{}, nil } +func (r *fakeLuckyGiftRepository) GetLuckyGiftDrawRewardState(context.Context, string, []string) (domain.DrawRewardState, error) { + return r.rewardState, nil +} + func (r *fakeLuckyGiftRepository) ClaimPendingLuckyGiftOutbox(context.Context, string, int64, time.Duration, int) ([]domain.DrawOutbox, error) { return r.outbox, nil } @@ -342,12 +477,21 @@ func (r *fakeLuckyGiftRepository) MarkLuckyGiftDrawsFailed(context.Context, stri } type fakeLuckyGiftWallet struct { - last *walletv1.CreditLuckyGiftRewardRequest + last *walletv1.CreditLuckyGiftRewardRequest + err error + balanceAfter int64 } func (w *fakeLuckyGiftWallet) CreditLuckyGiftReward(_ context.Context, req *walletv1.CreditLuckyGiftRewardRequest, _ ...grpc.CallOption) (*walletv1.CreditLuckyGiftRewardResponse, error) { w.last = req - return &walletv1.CreditLuckyGiftRewardResponse{TransactionId: "wallet_tx_lucky", Amount: req.GetAmount()}, nil + if w.err != nil { + return nil, w.err + } + return &walletv1.CreditLuckyGiftRewardResponse{ + TransactionId: "wallet_tx_lucky", + Amount: req.GetAmount(), + Balance: &walletv1.AssetBalance{AvailableAmount: w.balanceAfter}, + }, nil } type fakeLuckyGiftPublisher struct { diff --git a/services/activity-service/internal/storage/mysql/lucky_gift_repository.go b/services/activity-service/internal/storage/mysql/lucky_gift_repository.go index 55f971e3..224e48da 100644 --- a/services/activity-service/internal/storage/mysql/lucky_gift_repository.go +++ b/services/activity-service/internal/storage/mysql/lucky_gift_repository.go @@ -1055,6 +1055,12 @@ func luckyAggregateDrawResults(cmd domain.DrawCommand, results []domain.DrawResu } aggregate := results[0] aggregate.CommandID = cmd.CommandID + aggregate.DrawIDs = make([]string, 0, len(results)) + for _, result := range results { + if strings.TrimSpace(result.DrawID) != "" { + aggregate.DrawIDs = append(aggregate.DrawIDs, result.DrawID) + } + } if len(results) == 1 { return aggregate } @@ -1467,6 +1473,54 @@ func (r *Repository) MarkLuckyGiftDrawGranted(ctx context.Context, appCode strin return r.MarkLuckyGiftDrawsGranted(ctx, appCode, []string{drawID}, walletTransactionID, nowMS) } +func (r *Repository) GetLuckyGiftDrawRewardState(ctx context.Context, appCode string, drawIDs []string) (domain.DrawRewardState, error) { + if r == nil || r.db == nil { + return domain.DrawRewardState{}, xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + drawIDs = normalizeLuckyDrawIDs(drawIDs) + if len(drawIDs) == 0 { + return domain.DrawRewardState{}, nil + } + args := []any{appcode.Normalize(appCode)} + for _, drawID := range drawIDs { + args = append(args, drawID) + } + rows, err := r.db.QueryContext(ctx, ` + SELECT reward_status, COALESCE(reward_transaction_id, ''), COUNT(*) + FROM lucky_draw_records + WHERE app_code = ? AND draw_id IN (`+luckySQLPlaceholders(len(drawIDs))+`) + GROUP BY reward_status, reward_transaction_id`, args...) + if err != nil { + return domain.DrawRewardState{}, err + } + defer rows.Close() + var total int64 + var granted int64 + walletTransactionID := "" + for rows.Next() { + var status string + var txID string + var count int64 + if err := rows.Scan(&status, &txID, &count); err != nil { + return domain.DrawRewardState{}, err + } + total += count + if status == domain.StatusGranted { + granted += count + if walletTransactionID == "" { + walletTransactionID = txID + } + } + } + if err := rows.Err(); err != nil { + return domain.DrawRewardState{}, err + } + return domain.DrawRewardState{ + AllGranted: total == int64(len(drawIDs)) && granted == total, + WalletTransactionID: walletTransactionID, + }, nil +} + // MarkLuckyGiftDrawsGranted 是聚合 outbox 的终态收敛入口。 // 钱包只加币一次，但 draw_records 仍有多条；这里必须批量 granted，避免后台看到大量已发奖但仍 pending 的明细。 func (r *Repository) MarkLuckyGiftDrawsGranted(ctx context.Context, appCode string, drawIDs []string, walletTransactionID string, nowMS int64) error { @@ -1683,7 +1737,7 @@ func (r *Repository) ListLuckyGiftDraws(ctx context.Context, query domain.DrawQu COALESCE(JSON_UNQUOTE(JSON_EXTRACT(candidate_tiers_json, '$.multiplier_ppm')), '0'), base_reward_coins, room_atmosphere_reward_coins, activity_subsidy_coins, effective_reward_coins, COALESCE(CAST(budget_sources_json AS CHAR), ''), reward_status, - rtp_window_index, gift_rtp_window_index, stage_feedback, high_multiplier, created_at_ms + COALESCE(reward_transaction_id, ''), rtp_window_index, gift_rtp_window_index, stage_feedback, high_multiplier, created_at_ms FROM lucky_draw_records WHERE `+whereSQL+` ORDER BY created_at_ms DESC, draw_id DESC @@ -1698,10 +1752,11 @@ func (r *Repository) ListLuckyGiftDraws(ctx context.Context, query domain.DrawQu if err := rows.Scan(&item.DrawID, &item.CommandID, &item.PoolID, &item.GiftID, &item.RuleVersion, &item.ExperiencePool, &item.SelectedTierID, &item.MultiplierPPM, &item.BaseRewardCoins, &item.RoomAtmosphereRewardCoins, &item.ActivitySubsidyCoins, &item.EffectiveRewardCoins, - &item.BudgetSourcesJSON, &item.RewardStatus, &item.RTPWindowIndex, &item.GiftRTPWindowIndex, + &item.BudgetSourcesJSON, &item.RewardStatus, &item.WalletTransactionID, &item.RTPWindowIndex, &item.GiftRTPWindowIndex, &item.StageFeedback, &item.HighMultiplier, &item.CreatedAtMS); err != nil { return nil, 0, err } + item.DrawIDs = []string{item.DrawID} items = append(items, item) } return items, total, rows.Err() @@ -2505,7 +2560,7 @@ func (r *Repository) getLuckyDrawByCommand(ctx context.Context, tx *sql.Tx, appC COALESCE(JSON_UNQUOTE(JSON_EXTRACT(candidate_tiers_json, '$.multiplier_ppm')), '0'), base_reward_coins, room_atmosphere_reward_coins, activity_subsidy_coins, effective_reward_coins, COALESCE(CAST(budget_sources_json AS CHAR), ''), reward_status, - rtp_window_index, gift_rtp_window_index, stage_feedback, high_multiplier, created_at_ms + COALESCE(reward_transaction_id, ''), rtp_window_index, gift_rtp_window_index, stage_feedback, high_multiplier, created_at_ms FROM lucky_draw_records WHERE app_code = ? AND command_id = ?`+lockSQL, appCode, commandID, @@ -2514,13 +2569,14 @@ func (r *Repository) getLuckyDrawByCommand(ctx context.Context, tx *sql.Tx, appC if err := row.Scan(&item.DrawID, &item.CommandID, &item.PoolID, &item.GiftID, &item.RuleVersion, &item.ExperiencePool, &item.SelectedTierID, &item.MultiplierPPM, &item.BaseRewardCoins, &item.RoomAtmosphereRewardCoins, &item.ActivitySubsidyCoins, &item.EffectiveRewardCoins, - &item.BudgetSourcesJSON, &item.RewardStatus, &item.RTPWindowIndex, &item.GiftRTPWindowIndex, + &item.BudgetSourcesJSON, &item.RewardStatus, &item.WalletTransactionID, &item.RTPWindowIndex, &item.GiftRTPWindowIndex, &item.StageFeedback, &item.HighMultiplier, &item.CreatedAtMS); err != nil { if errors.Is(err, sql.ErrNoRows) { return domain.DrawResult{}, false, nil } return domain.DrawResult{}, false, err } + item.DrawIDs = []string{item.DrawID} return item, true, nil } diff --git a/services/activity-service/internal/transport/grpc/lucky_gift_server.go b/services/activity-service/internal/transport/grpc/lucky_gift_server.go index 86247214..362bc41d 100644 --- a/services/activity-service/internal/transport/grpc/lucky_gift_server.go +++ b/services/activity-service/internal/transport/grpc/lucky_gift_server.go @@ -265,6 +265,8 @@ func luckyDrawResultToProto(result domain.DrawResult) *activityv1.LuckyGiftDrawR StageFeedback: result.StageFeedback, HighMultiplier: result.HighMultiplier, CreatedAtMs: result.CreatedAtMS, + WalletTransactionId: result.WalletTransactionID, + CoinBalanceAfter: result.CoinBalanceAfter, } } diff --git a/services/gateway-service/internal/transport/http/response_test.go b/services/gateway-service/internal/transport/http/response_test.go index 228b2e27..c5ba937e 100644 --- a/services/gateway-service/internal/transport/http/response_test.go +++ b/services/gateway-service/internal/transport/http/response_test.go @@ -1720,7 +1720,9 @@ func TestSendGiftResponseIncludesLuckyGiftDraw(t *testing.T) { SelectedTierId: "novice_2x", MultiplierPpm: 2_000_000, EffectiveRewardCoins: 200, - RewardStatus: "pending", + RewardStatus: "granted", + WalletTransactionId: "wallet_tx_lucky", + CoinBalanceAfter: 8800, CreatedAtMs: 1_779_258_000_000, }, }} @@ -1742,13 +1744,16 @@ func TestSendGiftResponseIncludesLuckyGiftDraw(t *testing.T) { PoolID string `json:"pool_id"` MultiplierPPM int64 `json:"multiplier_ppm"` EffectiveRewardCoins int64 `json:"effective_reward_coins"` + RewardStatus string `json:"reward_status"` + WalletTransactionID string `json:"wallet_transaction_id"` + CoinBalanceAfter int64 `json:"coin_balance_after"` } `json:"lucky_gift"` } `json:"data"` } if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil { t.Fatalf("decode response failed: %v body=%s", err, recorder.Body.String()) } - if envelope.Code != "OK" || envelope.Data.LuckyGift.DrawID != "lucky_draw_test" || envelope.Data.LuckyGift.PoolID != "super_lucky" || envelope.Data.LuckyGift.MultiplierPPM != 2_000_000 || envelope.Data.LuckyGift.EffectiveRewardCoins != 200 { + if envelope.Code != "OK" || envelope.Data.LuckyGift.DrawID != "lucky_draw_test" || envelope.Data.LuckyGift.PoolID != "super_lucky" || envelope.Data.LuckyGift.MultiplierPPM != 2_000_000 || envelope.Data.LuckyGift.EffectiveRewardCoins != 200 || envelope.Data.LuckyGift.RewardStatus != "granted" || envelope.Data.LuckyGift.WalletTransactionID != "wallet_tx_lucky" || envelope.Data.LuckyGift.CoinBalanceAfter != 8800 { t.Fatalf("lucky_gift response mismatch: %+v", envelope.Data.LuckyGift) } } diff --git a/services/room-service/configs/config.docker.yaml b/services/room-service/configs/config.docker.yaml index f9fe076a..54be4d00 100644 --- a/services/room-service/configs/config.docker.yaml +++ b/services/room-service/configs/config.docker.yaml @@ -14,6 +14,7 @@ node_registry_ttl: "30s" node_registry_heartbeat_interval: "10s" owner_forward_timeout: "2s" lease_ttl: "10s" +lucky_gift_send_lock_ttl: "5s" rank_limit: 20 snapshot_every_n: 10 wallet_service_addr: "wallet-service:13004" diff --git a/services/room-service/configs/config.tencent.example.yaml b/services/room-service/configs/config.tencent.example.yaml index b3545fe3..405066f8 100644 --- a/services/room-service/configs/config.tencent.example.yaml +++ b/services/room-service/configs/config.tencent.example.yaml @@ -53,6 +53,7 @@ mysql_auto_migrate: false redis_addr: "TENCENT_REDIS_HOST:6379" redis_password: "TENCENT_REDIS_PASSWORD" redis_db: 0 +lucky_gift_send_lock_ttl: "5s" presence_stale_after: "2m" presence_stale_scan_interval: "30s" mic_publish_timeout: "15s" diff --git a/services/room-service/configs/config.yaml b/services/room-service/configs/config.yaml index c0d89dd6..c21d9718 100644 --- a/services/room-service/configs/config.yaml +++ b/services/room-service/configs/config.yaml @@ -8,6 +8,7 @@ log: include_response_body: false max_payload_bytes: 2048 lease_ttl: "10s" +lucky_gift_send_lock_ttl: "5s" rank_limit: 20 snapshot_every_n: 10 grpc_addr: ":13001" diff --git a/services/room-service/internal/app/app.go b/services/room-service/internal/app/app.go index f2265d38..3a32c120 100644 --- a/services/room-service/internal/app/app.go +++ b/services/room-service/internal/app/app.go @@ -205,6 +205,8 @@ func New(cfg config.Config) (*App, error) { RTCUserRemover: rtcUserRemover, RoomTreasureOpenScheduler: treasureOpenScheduler, RoomGiftLeaderboard: roomservice.NewRedisRoomGiftLeaderboardStore(redisClient), + LuckyGiftSendLocker: roomservice.NewRedisLuckyGiftSendLocker(redisClient), + LuckyGiftSendLockTTL: cfg.LuckyGiftSendLockTTL, }, directory, repository, integration.NewGRPCWalletClient(walletConn), roomPublisher, outboxPublisher, luckyGiftClient) if cfg.RocketMQ.TreasureOpen.Enabled { treasureConsumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.TreasureOpen.ConsumerGroup, cfg.RocketMQ.TreasureOpen.ConsumerMaxReconsumeTimes)) diff --git a/services/room-service/internal/config/config.go b/services/room-service/internal/config/config.go index eb7801ed..616d2cf9 100644 --- a/services/room-service/internal/config/config.go +++ b/services/room-service/internal/config/config.go @@ -42,6 +42,8 @@ type Config struct { OwnerForwardTimeout time.Duration `yaml:"owner_forward_timeout"` // LeaseTTL 是 room owner 租约有效期，节点故障后必须等它过期才能接管。 LeaseTTL time.Duration `yaml:"lease_ttl"` + // LuckyGiftSendLockTTL 是同一用户幸运礼物扣费、抽奖、同步返奖链路的 Redis 短锁 TTL。 + LuckyGiftSendLockTTL time.Duration `yaml:"lucky_gift_send_lock_ttl"` // RankLimit 是房间本地礼物榜 top N。 RankLimit int `yaml:"rank_limit"` // SnapshotEveryN 控制按房间版本保存快照的频率。 @@ -220,19 +222,20 @@ func (c TencentIMConfig) RESTConfig() tencentim.RESTConfig { func Default() Config { // 默认端口遵守项目 13xxx 约束，MySQL/Redis 指向本地 Docker 暴露端口。 return Config{ - ServiceName: "room-service", - Environment: "local", - NodeID: "room-node-local", - GRPCAddr: ":13001", - AdvertiseAddr: "127.0.0.1:13001", - HealthHTTPAddr: ":13101", - NodeRegistryTTL: 30 * time.Second, - OwnerForwardTimeout: 2 * time.Second, - LeaseTTL: 10 * time.Second, - RankLimit: 20, - SnapshotEveryN: 10, - WalletServiceAddr: "127.0.0.1:13004", - ActivityServiceAddr: "127.0.0.1:13006", + ServiceName: "room-service", + Environment: "local", + NodeID: "room-node-local", + GRPCAddr: ":13001", + AdvertiseAddr: "127.0.0.1:13001", + HealthHTTPAddr: ":13101", + NodeRegistryTTL: 30 * time.Second, + OwnerForwardTimeout: 2 * time.Second, + LeaseTTL: 10 * time.Second, + LuckyGiftSendLockTTL: 5 * time.Second, + RankLimit: 20, + SnapshotEveryN: 10, + WalletServiceAddr: "127.0.0.1:13004", + ActivityServiceAddr: "127.0.0.1:13006", TencentIM: TencentIMConfig{ Enabled: false, AdminIdentifier: "administrator", @@ -348,6 +351,9 @@ func Normalize(cfg Config) (Config, error) { if cfg.OwnerForwardTimeout <= 0 { cfg.OwnerForwardTimeout = 2 * time.Second } + if cfg.LuckyGiftSendLockTTL <= 0 { + cfg.LuckyGiftSendLockTTL = 5 * time.Second + } cfg.WalletServiceAddr = strings.TrimSpace(cfg.WalletServiceAddr) if cfg.WalletServiceAddr == "" { cfg.WalletServiceAddr = "127.0.0.1:13004" diff --git a/services/room-service/internal/room/service/gift.go b/services/room-service/internal/room/service/gift.go index 94eed286..cea95ea4 100644 --- a/services/room-service/internal/room/service/gift.go +++ b/services/room-service/internal/room/service/gift.go @@ -98,74 +98,82 @@ func (s *Service) SendGift(ctx context.Context, req *roomv1.SendGiftRequest) (*r return mutationResult{}, nil, err } - // 钱包扣费在房间状态变更前完成；失败时不写 command log、不进入 Room Cell 提交态。 - walletStartedAt := time.Now() - billing, err := s.wallet.DebitGift(ctx, &walletv1.DebitGiftRequest{ - CommandId: cmd.ID(), - RoomId: cmd.RoomID(), - SenderUserId: cmd.ActorUserID(), - TargetUserId: cmd.TargetUserID, - GiftId: cmd.GiftID, - GiftCount: cmd.GiftCount, - AppCode: appcode.FromContext(ctx), - RegionId: roomMeta.VisibleRegionID, - TargetIsHost: cmd.TargetIsHost, - TargetHostRegionId: cmd.TargetHostRegionID, - TargetAgencyOwnerUserId: cmd.TargetAgencyOwnerUserID, - }) - walletDebitMS := elapsedMS(walletStartedAt) - if err != nil { + var billing *walletv1.DebitGiftResponse + var walletDebitMS int64 + settledCommand := cmd + var luckyGift *roomv1.LuckyGiftDrawResult + if err := s.withLuckyGiftSendLock(ctx, luckyEnabled, cmd.ActorUserID(), func() error { + // 钱包扣费在房间状态变更前完成；失败时不写 command log、不进入 Room Cell 提交态。 + walletStartedAt := time.Now() + var err error + billing, err = s.wallet.DebitGift(ctx, &walletv1.DebitGiftRequest{ + CommandId: cmd.ID(), + RoomId: cmd.RoomID(), + SenderUserId: cmd.ActorUserID(), + TargetUserId: cmd.TargetUserID, + GiftId: cmd.GiftID, + GiftCount: cmd.GiftCount, + AppCode: appcode.FromContext(ctx), + RegionId: roomMeta.VisibleRegionID, + TargetIsHost: cmd.TargetIsHost, + TargetHostRegionId: cmd.TargetHostRegionID, + TargetAgencyOwnerUserId: cmd.TargetAgencyOwnerUserID, + }) + walletDebitMS = elapsedMS(walletStartedAt) + if err != nil { + return err + } + if billing == nil { + return xerr.New(xerr.Unavailable, "wallet debit response is empty") + } + settledCommand.BillingReceiptID = billing.GetBillingReceiptId() + settledCommand.CoinSpent = billing.GetCoinSpent() + settledCommand.GiftPointAdded = billing.GetGiftPointAdded() + settledCommand.HeatValue = billing.GetHeatValue() + settledCommand.PriceVersion = billing.GetPriceVersion() + settledCommand.GiftTypeCode = billing.GetGiftTypeCode() + settledCommand.HostPeriodDiamondAdded = billing.GetHostPeriodDiamondAdded() + settledCommand.HostPeriodCycleKey = billing.GetHostPeriodCycleKey() + if !luckyEnabled { + // 未显式传 pool_id 的旧客户端仍可通过钱包礼物类型触发默认幸运礼物逻辑；新客户端应传明确 pool_id。 + luckyEnabled = s.shouldDrawLuckyGift(cmd.PoolID, billing.GetGiftTypeCode()) + } + if luckyEnabled { + // 抽奖必须发生在扣费成功之后；activity-service 只接收钱包结算后的 coin_spent，不信任客户端价格。 + drawResp, err := s.luckyGift.ExecuteLuckyGiftDraw(ctx, &activityv1.ExecuteLuckyGiftDrawRequest{ + LuckyGift: &activityv1.LuckyGiftMeta{ + Meta: activityMetaFromRoom(ctx, req.GetMeta()), + CommandId: cmd.ID(), + UserId: cmd.ActorUserID(), + TargetUserId: cmd.TargetUserID, + // 目前没有独立设备 ID 字段，优先用 session_id；没有时退回 command_id，保证风控 scope 不为空。 + DeviceId: luckyGiftDeviceID(cmd), + RoomId: cmd.RoomID(), + AnchorId: luckyGiftAnchorID(roomMeta), + GiftId: cmd.GiftID, + GiftCount: cmd.GiftCount, + CoinSpent: billing.GetCoinSpent(), + // 房间链路统一使用 UTC epoch ms；不能把本地时区时间传给活动风控窗口。 + PaidAtMs: now.UTC().UnixMilli(), + PoolId: cmd.PoolID, + VisibleRegionId: roomMeta.VisibleRegionID, + }, + }) + if err != nil { + // 抽奖失败时整笔送礼失败并依赖 wallet 幂等重试；不能落普通礼物后丢失幸运礼物抽奖事实。 + return err + } + if drawResp == nil { + return xerr.New(xerr.Unavailable, "lucky gift draw response is empty") + } + // 同步返回只给当前送礼用户做即时表现；金币余额以 activity/wallet 快路径回执为准，房间开奖结果仍由 activity outbox 补偿。 + luckyGift = luckyGiftResultFromProto(drawResp.GetResult()) + } + return nil + }); err != nil { return mutationResult{}, nil, err } - if billing == nil { - return mutationResult{}, nil, xerr.New(xerr.Unavailable, "wallet debit response is empty") - } heatValue := billing.GetHeatValue() - settledCommand := cmd - settledCommand.BillingReceiptID = billing.GetBillingReceiptId() - settledCommand.CoinSpent = billing.GetCoinSpent() - settledCommand.GiftPointAdded = billing.GetGiftPointAdded() - settledCommand.HeatValue = heatValue - settledCommand.PriceVersion = billing.GetPriceVersion() - settledCommand.GiftTypeCode = billing.GetGiftTypeCode() - settledCommand.HostPeriodDiamondAdded = billing.GetHostPeriodDiamondAdded() - settledCommand.HostPeriodCycleKey = billing.GetHostPeriodCycleKey() - if !luckyEnabled { - // 未显式传 pool_id 的旧客户端仍可通过钱包礼物类型触发默认幸运礼物逻辑；新客户端应传明确 pool_id。 - luckyEnabled = s.shouldDrawLuckyGift(cmd.PoolID, billing.GetGiftTypeCode()) - } - var luckyGift *roomv1.LuckyGiftDrawResult - if luckyEnabled { - // 抽奖必须发生在扣费成功之后；activity-service 只接收钱包结算后的 coin_spent，不信任客户端价格。 - drawResp, err := s.luckyGift.ExecuteLuckyGiftDraw(ctx, &activityv1.ExecuteLuckyGiftDrawRequest{ - LuckyGift: &activityv1.LuckyGiftMeta{ - Meta: activityMetaFromRoom(ctx, req.GetMeta()), - CommandId: cmd.ID(), - UserId: cmd.ActorUserID(), - TargetUserId: cmd.TargetUserID, - // 目前没有独立设备 ID 字段，优先用 session_id；没有时退回 command_id，保证风控 scope 不为空。 - DeviceId: luckyGiftDeviceID(cmd), - RoomId: cmd.RoomID(), - AnchorId: luckyGiftAnchorID(roomMeta), - GiftId: cmd.GiftID, - GiftCount: cmd.GiftCount, - CoinSpent: billing.GetCoinSpent(), - // 房间链路统一使用 UTC epoch ms；不能把本地时区时间传给活动风控窗口。 - PaidAtMs: now.UTC().UnixMilli(), - PoolId: cmd.PoolID, - VisibleRegionId: roomMeta.VisibleRegionID, - }, - }) - if err != nil { - // 抽奖失败时整笔送礼失败并依赖 wallet 幂等重试；不能落普通礼物后丢失幸运礼物抽奖事实。 - return mutationResult{}, nil, err - } - if drawResp == nil { - return mutationResult{}, nil, xerr.New(xerr.Unavailable, "lucky gift draw response is empty") - } - // 同步返回只给当前送礼用户做即时表现；返奖到账和房间开奖结果仍由 activity outbox 补偿。 - luckyGift = luckyGiftResultFromProto(drawResp.GetResult()) - } // 扣费成功后，Room Cell 同步更新热度和本地礼物榜。 current.Heat += heatValue @@ -314,6 +322,24 @@ func (s *Service) shouldDrawLuckyGift(poolID string, giftTypeCode string) bool { } } +func (s *Service) withLuckyGiftSendLock(ctx context.Context, enabled bool, userID int64, fn func() error) error { + if !enabled || s == nil || s.luckyGiftSendLocker == nil { + return fn() + } + release, err := s.luckyGiftSendLocker.AcquireLuckyGiftSendLock(ctx, appcode.FromContext(ctx), userID, s.luckyGiftSendLockTTL) + if err != nil { + return err + } + if release != nil { + defer func() { + releaseCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + release(releaseCtx) + }() + } + return fn() +} + func activityMetaFromRoom(ctx context.Context, meta *roomv1.RequestMeta) *activityv1.RequestMeta { return &activityv1.RequestMeta{ RequestId: meta.GetRequestId(), @@ -360,6 +386,8 @@ func luckyGiftResultFromProto(result *activityv1.LuckyGiftDrawResult) *roomv1.Lu StageFeedback: result.GetStageFeedback(), HighMultiplier: result.GetHighMultiplier(), CreatedAtMs: result.GetCreatedAtMs(), + WalletTransactionId: result.GetWalletTransactionId(), + CoinBalanceAfter: result.GetCoinBalanceAfter(), } } diff --git a/services/room-service/internal/room/service/lucky_gift_send_lock.go b/services/room-service/internal/room/service/lucky_gift_send_lock.go new file mode 100644 index 00000000..dfdf5f9a --- /dev/null +++ b/services/room-service/internal/room/service/lucky_gift_send_lock.go @@ -0,0 +1,60 @@ +package service + +import ( + "context" + "fmt" + "strconv" + "time" + + "github.com/redis/go-redis/v9" + "hyapp/pkg/appcode" + "hyapp/pkg/idgen" + "hyapp/pkg/xerr" +) + +const releaseLuckyGiftSendLockScript = ` +if redis.call("GET", KEYS[1]) == ARGV[1] then + return redis.call("DEL", KEYS[1]) +end +return 0 +` + +type LuckyGiftSendLocker interface { + AcquireLuckyGiftSendLock(ctx context.Context, appCode string, userID int64, ttl time.Duration) (func(context.Context), error) +} + +type RedisLuckyGiftSendLocker struct { + client *redis.Client +} + +func NewRedisLuckyGiftSendLocker(client *redis.Client) *RedisLuckyGiftSendLocker { + return &RedisLuckyGiftSendLocker{client: client} +} + +func (l *RedisLuckyGiftSendLocker) AcquireLuckyGiftSendLock(ctx context.Context, appCode string, userID int64, ttl time.Duration) (func(context.Context), error) { + if l == nil || l.client == nil { + return nil, nil + } + if userID <= 0 { + return nil, xerr.New(xerr.InvalidArgument, "lucky gift sender is invalid") + } + if ttl <= 0 { + ttl = 5 * time.Second + } + key := luckyGiftSendLockKey(appCode, userID) + token := idgen.New("lucky_gift_send_lock") + ok, err := l.client.SetNX(ctx, key, token, ttl).Result() + if err != nil { + return nil, err + } + if !ok { + return nil, xerr.New(xerr.Conflict, "lucky gift send is already processing") + } + return func(releaseCtx context.Context) { + _ = l.client.Eval(releaseCtx, releaseLuckyGiftSendLockScript, []string{key}, token).Err() + }, nil +} + +func luckyGiftSendLockKey(rawAppCode string, userID int64) string { + return fmt.Sprintf("lucky_gift_send:%s:%s", appcode.Normalize(rawAppCode), strconv.FormatInt(userID, 10)) +} diff --git a/services/room-service/internal/room/service/lucky_gift_send_lock_test.go b/services/room-service/internal/room/service/lucky_gift_send_lock_test.go new file mode 100644 index 00000000..e2e62983 --- /dev/null +++ b/services/room-service/internal/room/service/lucky_gift_send_lock_test.go @@ -0,0 +1,59 @@ +package service + +import ( + "context" + "testing" + "time" + + "hyapp/pkg/appcode" +) + +func TestWithLuckyGiftSendLockWrapsLuckyPath(t *testing.T) { + locker := &fakeLuckyGiftSendLocker{} + svc := &Service{luckyGiftSendLocker: locker, luckyGiftSendLockTTL: 3 * time.Second} + called := false + + err := svc.withLuckyGiftSendLock(appcode.WithContext(context.Background(), "lalu"), true, 42, func() error { + if !locker.acquired || locker.released { + t.Fatalf("lock must be held while lucky gift send runs: %+v", locker) + } + called = true + return nil + }) + if err != nil { + t.Fatalf("withLuckyGiftSendLock failed: %v", err) + } + if !called || !locker.acquired || !locker.released || locker.appCode != "lalu" || locker.userID != 42 || locker.ttl != 3*time.Second { + t.Fatalf("lock lifecycle mismatch: called=%v locker=%+v", called, locker) + } +} + +func TestWithLuckyGiftSendLockBypassesNormalGift(t *testing.T) { + locker := &fakeLuckyGiftSendLocker{} + svc := &Service{luckyGiftSendLocker: locker, luckyGiftSendLockTTL: 3 * time.Second} + + if err := svc.withLuckyGiftSendLock(context.Background(), false, 42, func() error { return nil }); err != nil { + t.Fatalf("normal gift path should bypass lock: %v", err) + } + if locker.acquired { + t.Fatalf("normal gift path must not acquire lucky gift lock") + } +} + +type fakeLuckyGiftSendLocker struct { + acquired bool + released bool + appCode string + userID int64 + ttl time.Duration +} + +func (l *fakeLuckyGiftSendLocker) AcquireLuckyGiftSendLock(_ context.Context, appCode string, userID int64, ttl time.Duration) (func(context.Context), error) { + l.acquired = true + l.appCode = appCode + l.userID = userID + l.ttl = ttl + return func(context.Context) { + l.released = true + }, nil +} diff --git a/services/room-service/internal/room/service/lucky_gift_test.go b/services/room-service/internal/room/service/lucky_gift_test.go index f3feff88..2573f91f 100644 --- a/services/room-service/internal/room/service/lucky_gift_test.go +++ b/services/room-service/internal/room/service/lucky_gift_test.go @@ -46,7 +46,9 @@ func (c *luckyGiftTestClient) ExecuteLuckyGiftDraw(_ context.Context, req *activ MultiplierPpm: 2_000_000, BaseRewardCoins: 200, EffectiveRewardCoins: 200, - RewardStatus: "pending", + RewardStatus: "granted", + WalletTransactionId: "wallet_tx_lucky", + CoinBalanceAfter: 8800, CreatedAtMs: 1_779_258_000_000, }}, nil } @@ -94,7 +96,7 @@ func TestSendGiftReturnsLuckyGiftDrawResult(t *testing.T) { if err != nil { t.Fatalf("send lucky gift failed: %v", err) } - if resp.GetLuckyGift().GetDrawId() != "lucky_draw_test" || resp.GetLuckyGift().GetMultiplierPpm() != 2_000_000 || resp.GetLuckyGift().GetEffectiveRewardCoins() != 200 { + if resp.GetLuckyGift().GetDrawId() != "lucky_draw_test" || resp.GetLuckyGift().GetMultiplierPpm() != 2_000_000 || resp.GetLuckyGift().GetEffectiveRewardCoins() != 200 || resp.GetLuckyGift().GetRewardStatus() != "granted" || resp.GetLuckyGift().GetWalletTransactionId() != "wallet_tx_lucky" || resp.GetLuckyGift().GetCoinBalanceAfter() != 8800 { t.Fatalf("lucky gift result mismatch: %+v", resp.GetLuckyGift()) } if len(luckyGift.checks) != 1 || luckyGift.checks[0].GetPoolId() != "super_lucky" || luckyGift.checks[0].GetGiftId() != "rose" { diff --git a/services/room-service/internal/room/service/service.go b/services/room-service/internal/room/service/service.go index ebc979f2..c136ecd3 100644 --- a/services/room-service/internal/room/service/service.go +++ b/services/room-service/internal/room/service/service.go @@ -36,6 +36,10 @@ type Config struct { RoomTreasureOpenScheduler integration.RoomTreasureOpenScheduler // RoomGiftLeaderboard 是跨房间金币榜读模型，SendGift 提交后 best-effort 写入。 RoomGiftLeaderboard RoomGiftLeaderboardStore + // LuckyGiftSendLocker 串行化同一用户的幸运礼物扣费、抽奖和同步返奖链路。 + LuckyGiftSendLocker LuckyGiftSendLocker + // LuckyGiftSendLockTTL 是幸运礼物短锁过期时间。 + LuckyGiftSendLockTTL time.Duration // Clock 允许测试注入稳定时间，生产默认使用系统时钟。 Clock clock.Clock } @@ -72,6 +76,10 @@ type Service struct { roomTreasureOpenScheduler integration.RoomTreasureOpenScheduler // roomGiftLeaderboard 只保存房间维度金币消耗 zset，不承载 Room Cell 核心状态。 roomGiftLeaderboard RoomGiftLeaderboardStore + // luckyGiftSendLocker 只保护同一用户幸运礼物扣费到同步返奖的小事务链路。 + luckyGiftSendLocker LuckyGiftSendLocker + // luckyGiftSendLockTTL 控制 Redis 短锁自动释放窗口。 + luckyGiftSendLockTTL time.Duration // rtcUserRemover 在 Room Cell 提交驱逐后同步移除实时音频房连接。 rtcUserRemover integration.RTCUserRemover // draining 表示当前节点正在下线，只允许已经进入执行链路的命令收尾。 @@ -159,6 +167,10 @@ func New(cfg Config, directory router.Directory, repository Repository, wallet i // MicUp 只是业务占麦，客户端必须在短窗口内确认 RTC 发流成功。 micPublishTimeout = 15 * time.Second } + luckyGiftSendLockTTL := cfg.LuckyGiftSendLockTTL + if luckyGiftSendLockTTL <= 0 { + luckyGiftSendLockTTL = 5 * time.Second + } var luckyGiftClient integration.LuckyGiftClient if len(luckyGift) > 0 { @@ -181,6 +193,8 @@ func New(cfg Config, directory router.Directory, repository Repository, wallet i outboxPublisher: outboxPublisher, roomTreasureOpenScheduler: cfg.RoomTreasureOpenScheduler, roomGiftLeaderboard: cfg.RoomGiftLeaderboard, + luckyGiftSendLocker: cfg.LuckyGiftSendLocker, + luckyGiftSendLockTTL: luckyGiftSendLockTTL, rtcUserRemover: cfg.RTCUserRemover, cells: make(map[string]*cell.RoomCell), }