fix: credit lucky gift rewards synchronously
This commit is contained in:
zhx
parent
fd215eadb6
commit
afc62eca1e
@ -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" +
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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" +
|
||||
|
||||
@ -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 是后台配置给客户端展示的宝箱奖励候选项。
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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 == "" {
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
|
||||
@ -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,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -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))
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -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(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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))
|
||||
}
|
||||
@ -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
|
||||
}
|
||||
@ -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" {
|
||||
|
||||
@ -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),
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user