303 lines
13 KiB
Go
303 lines
13 KiB
Go
package service_test
|
||
|
||
import (
|
||
"context"
|
||
"sync/atomic"
|
||
"testing"
|
||
"time"
|
||
|
||
luckygiftv1 "hyapp.local/api/proto/luckygift/v1"
|
||
roomv1 "hyapp.local/api/proto/room/v1"
|
||
walletv1 "hyapp.local/api/proto/wallet/v1"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
"hyapp/services/room-service/internal/integration"
|
||
roomservice "hyapp/services/room-service/internal/room/service"
|
||
"hyapp/services/room-service/internal/router"
|
||
"hyapp/services/room-service/internal/testutil/mysqltest"
|
||
)
|
||
|
||
type concurrentV2Wallet struct {
|
||
followTestWallet
|
||
calls atomic.Int32
|
||
started chan struct{}
|
||
release chan struct{}
|
||
walletPaidAtMS int64
|
||
omitPaidAt bool
|
||
}
|
||
|
||
func (w *concurrentV2Wallet) DebitGift(ctx context.Context, _ *walletv1.DebitGiftRequest) (*walletv1.DebitGiftResponse, error) {
|
||
if w.calls.Add(1) == 1 && w.started != nil {
|
||
close(w.started)
|
||
}
|
||
if w.release != nil {
|
||
select {
|
||
case <-ctx.Done():
|
||
return nil, ctx.Err()
|
||
case <-w.release:
|
||
}
|
||
}
|
||
paidAtMS := w.walletPaidAtMS
|
||
if paidAtMS <= 0 && !w.omitPaidAt {
|
||
paidAtMS = 1_700_000_000_000
|
||
}
|
||
return &walletv1.DebitGiftResponse{
|
||
BillingReceiptId: "receipt-v2-stable",
|
||
TransactionId: "wallet-tx-v2-stable",
|
||
CoinSpent: 100,
|
||
ChargeAmount: 100,
|
||
HeatValue: 100,
|
||
BalanceAfter: 900,
|
||
BalanceVersion: 17,
|
||
GiftIncomeBalanceAfter: 25,
|
||
GiftTypeCode: "normal",
|
||
PaidAtMs: paidAtMS,
|
||
}, nil
|
||
}
|
||
|
||
func TestDynamicLuckyGiftRejectsMissingWalletPaidAtAfterDebit(t *testing.T) {
|
||
ctx := appcode.WithContext(context.Background(), appcode.Default)
|
||
repository := mysqltest.NewRepository(t)
|
||
wallet := &concurrentV2Wallet{omitPaidAt: true}
|
||
lucky := &recoverableLuckyClient{}
|
||
svc := roomservice.New(roomservice.Config{
|
||
NodeID: "node-gift-v2-missing-paid-at", LeaseTTL: 10 * time.Second, RankLimit: 20, SnapshotEveryN: 1,
|
||
}, router.NewMemoryDirectory(), repository, wallet, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher(), lucky)
|
||
|
||
roomID := "room-gift-v2-missing-paid-at"
|
||
createRocketRoom(t, ctx, svc, roomID, 351, 9001)
|
||
joinRocketRoom(t, ctx, svc, roomID, 352)
|
||
request := &roomv1.SendGiftRequest{
|
||
Meta: &roomv1.RequestMeta{CommandId: "cmd-v2-missing-paid-at", ActorUserId: 351, RoomId: roomID, AppCode: appcode.Default, DeviceId: "device-v2-missing-paid-at"},
|
||
TargetType: "user", TargetUserId: 352, GiftId: "lucky-rose", GiftCount: 1, PoolId: "lucky-pool",
|
||
}
|
||
if _, err := svc.SendGift(ctx, request); !xerr.IsCode(err, xerr.Unavailable) {
|
||
t.Fatalf("dynamic_v3 missing wallet paid_at_ms should fail closed, got %v", err)
|
||
}
|
||
if lucky.calls.Load() != 0 {
|
||
t.Fatalf("dynamic_v3 missing wallet paid_at_ms reached lucky random path %d times", lucky.calls.Load())
|
||
}
|
||
operation, exists, err := repository.GetGiftOperation(ctx, roomID, request.GetMeta().GetCommandId())
|
||
if err != nil || !exists || operation.Status != roomservice.GiftOperationStatusDebited {
|
||
t.Fatalf("already debited saga must remain recoverable: exists=%v operation=%+v err=%v", exists, operation, err)
|
||
}
|
||
}
|
||
|
||
func TestSendGiftConcurrentCommandReplaysOriginalV2AndKeepsV1Fields(t *testing.T) {
|
||
ctx := appcode.WithContext(context.Background(), appcode.Default)
|
||
repository := mysqltest.NewRepository(t)
|
||
wallet := &concurrentV2Wallet{started: make(chan struct{}), release: make(chan struct{})}
|
||
svc := roomservice.New(roomservice.Config{
|
||
NodeID: "node-gift-v2-race",
|
||
LeaseTTL: 10 * time.Second,
|
||
RankLimit: 20,
|
||
SnapshotEveryN: 1,
|
||
}, router.NewMemoryDirectory(), repository, wallet, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher())
|
||
|
||
roomID := "room-gift-v2-race"
|
||
senderID, targetID := int64(101), int64(202)
|
||
createRocketRoom(t, ctx, svc, roomID, senderID, 9001)
|
||
joinRocketRoom(t, ctx, svc, roomID, targetID)
|
||
req := &roomv1.SendGiftRequest{
|
||
Meta: &roomv1.RequestMeta{RequestId: "req-v2-race", CommandId: "combo-v2-race:1", ActorUserId: senderID, RoomId: roomID, AppCode: appcode.Default},
|
||
TargetType: "user", TargetUserId: targetID, GiftId: "rose", GiftCount: 5,
|
||
ComboSessionId: "combo-v2-race", BatchSeq: 1,
|
||
}
|
||
|
||
responses := make(chan *roomv1.SendGiftResponse, 2)
|
||
errs := make(chan error, 2)
|
||
go func() {
|
||
resp, err := svc.SendGift(ctx, req)
|
||
responses <- resp
|
||
errs <- err
|
||
}()
|
||
<-wallet.started
|
||
go func() {
|
||
resp, err := svc.SendGift(ctx, req)
|
||
responses <- resp
|
||
errs <- err
|
||
}()
|
||
close(wallet.release)
|
||
|
||
var got []*roomv1.SendGiftResponse
|
||
for range 2 {
|
||
if err := <-errs; err != nil {
|
||
t.Fatalf("concurrent SendGift failed: %v", err)
|
||
}
|
||
got = append(got, <-responses)
|
||
}
|
||
if calls := wallet.calls.Load(); calls != 1 {
|
||
t.Fatalf("same command_id must debit once, got %d wallet calls", calls)
|
||
}
|
||
replays := 0
|
||
for _, resp := range got {
|
||
if resp.GetV2() == nil {
|
||
t.Fatalf("V2 result is missing: %+v", resp)
|
||
}
|
||
if resp.GetV2().GetIdempotentReplay() {
|
||
replays++
|
||
}
|
||
if resp.GetV2().GetBillingReceiptId() != "receipt-v2-stable" || resp.GetV2().GetCoinBalanceAfter() != 900 || resp.GetV2().GetBalanceVersion() != 17 || resp.GetV2().GetSettledGiftCount() != 5 || resp.GetV2().GetCommittedRoomVersion() <= 0 {
|
||
t.Fatalf("V2 stable result mismatch: %+v", resp.GetV2())
|
||
}
|
||
// 旧 Flutter 仍读取顶层字段；首次响应和幂等重放都必须回填相同回执、余额和热度。
|
||
if resp.GetBillingReceiptId() != resp.GetV2().GetBillingReceiptId() || resp.GetCoinBalanceAfter() != resp.GetV2().GetCoinBalanceAfter() || resp.GetRoomHeat() != resp.GetV2().GetRoomHeat() {
|
||
t.Fatalf("V1 compatibility fields drifted: response=%+v v2=%+v", resp, resp.GetV2())
|
||
}
|
||
}
|
||
if replays != 1 {
|
||
t.Fatalf("exactly one overlapping request should be replayed, got %d", replays)
|
||
}
|
||
if heat := got[0].GetRoom().GetHeat(); heat != 100 {
|
||
t.Fatalf("same command_id changed room heat more than once: %d", heat)
|
||
}
|
||
}
|
||
|
||
type recoverableLuckyClient struct {
|
||
calls atomic.Int32
|
||
paidAtMS [2]atomic.Int64
|
||
}
|
||
|
||
func (*recoverableLuckyClient) CheckLuckyGift(context.Context, *luckygiftv1.CheckLuckyGiftRequest) (*luckygiftv1.CheckLuckyGiftResponse, error) {
|
||
return &luckygiftv1.CheckLuckyGiftResponse{Enabled: true, RuleVersion: 1, PoolId: "lucky-pool", StrategyVersion: "dynamic_v3"}, nil
|
||
}
|
||
|
||
func (c *recoverableLuckyClient) ExecuteLuckyGiftDraw(_ context.Context, req *luckygiftv1.ExecuteLuckyGiftDrawRequest) (*luckygiftv1.ExecuteLuckyGiftDrawResponse, error) {
|
||
meta := req.GetLuckyGift()
|
||
call := c.calls.Add(1)
|
||
if call <= 2 {
|
||
c.paidAtMS[call-1].Store(meta.GetPaidAtMs())
|
||
}
|
||
if call == 1 {
|
||
return nil, xerr.New(xerr.Unavailable, "temporary lucky outage")
|
||
}
|
||
return &luckygiftv1.ExecuteLuckyGiftDrawResponse{Result: &luckygiftv1.LuckyGiftDrawResult{
|
||
DrawId: "draw-recovered", CommandId: meta.GetCommandId(), GiftId: meta.GetGiftId(), PoolId: meta.GetPoolId(),
|
||
MultiplierPpm: 2_000_000, EffectiveRewardCoins: 200, RewardStatus: "granted", CoinBalanceAfter: 1100,
|
||
Hits: []*luckygiftv1.LuckyGiftHit{{GiftIndex: 1, DrawId: "draw-recovered", SelectedTierId: "2x", MultiplierPpm: 2_000_000, RewardCoins: 200}},
|
||
}}, nil
|
||
}
|
||
|
||
func (c *recoverableLuckyClient) BatchExecuteLuckyGiftDraw(ctx context.Context, req *luckygiftv1.BatchExecuteLuckyGiftDrawRequest) (*luckygiftv1.BatchExecuteLuckyGiftDrawResponse, error) {
|
||
results := make([]*luckygiftv1.LuckyGiftDrawResult, 0, len(req.GetLuckyGifts()))
|
||
for _, meta := range req.GetLuckyGifts() {
|
||
resp, err := c.ExecuteLuckyGiftDraw(ctx, &luckygiftv1.ExecuteLuckyGiftDrawRequest{LuckyGift: meta})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
results = append(results, resp.GetResult())
|
||
}
|
||
return &luckygiftv1.BatchExecuteLuckyGiftDrawResponse{Results: results}, nil
|
||
}
|
||
|
||
func TestGiftOperationWorkerFinishesDebitedRequestWithoutClientRetry(t *testing.T) {
|
||
ctx := appcode.WithContext(context.Background(), appcode.Default)
|
||
repository := mysqltest.NewRepository(t)
|
||
walletPaidAtMS := time.Date(2026, 7, 12, 23, 59, 59, 700_000_000, time.UTC).UnixMilli()
|
||
wallet := &concurrentV2Wallet{walletPaidAtMS: walletPaidAtMS}
|
||
lucky := &recoverableLuckyClient{}
|
||
clock := &fixedRoomRocketClock{now: time.Date(2026, 7, 12, 23, 59, 59, 800_000_000, time.UTC)}
|
||
svc := roomservice.New(roomservice.Config{
|
||
NodeID: "node-gift-v2-recovery",
|
||
LeaseTTL: 48 * time.Hour,
|
||
RankLimit: 20,
|
||
SnapshotEveryN: 1,
|
||
Clock: clock,
|
||
}, router.NewMemoryDirectory(), repository, wallet, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher(), lucky)
|
||
|
||
roomID := "room-gift-v2-recovery"
|
||
senderID, targetID := int64(301), int64(302)
|
||
createRocketRoom(t, ctx, svc, roomID, senderID, 9001)
|
||
joinRocketRoom(t, ctx, svc, roomID, targetID)
|
||
req := &roomv1.SendGiftRequest{
|
||
Meta: &roomv1.RequestMeta{RequestId: "req-v2-recovery", CommandId: "cmd-v2-recovery", ActorUserId: senderID, RoomId: roomID, AppCode: appcode.Default, DeviceId: "device-v2-recovery"},
|
||
TargetType: "user", TargetUserId: targetID, GiftId: "lucky-rose", GiftCount: 1, PoolId: "lucky-pool",
|
||
}
|
||
if _, err := svc.SendGift(ctx, req); !xerr.IsCode(err, xerr.Unavailable) {
|
||
t.Fatalf("first draw should fail after debit, got %v", err)
|
||
}
|
||
operation, exists, err := repository.GetGiftOperation(ctx, roomID, req.GetMeta().GetCommandId())
|
||
if err != nil || !exists || operation.Status != roomservice.GiftOperationStatusDebited {
|
||
t.Fatalf("debited saga was not persisted: exists=%v operation=%+v err=%v", exists, operation, err)
|
||
}
|
||
// 恢复跨过五分钟加权窗和 UTC 切日后，lucky-service 仍必须收到 wallet 首次交易时间；
|
||
// saga created_at 与 wallet paid_at 即使只差 100ms 也不是同一个 owner 事实，不能相互替代。
|
||
clock.now = clock.now.Add(6 * time.Minute)
|
||
if clock.Now().Sub(time.UnixMilli(operation.CreatedAtMS)) <= 5*time.Minute || clock.Now().UTC().Day() == time.UnixMilli(operation.CreatedAtMS).UTC().Day() {
|
||
t.Fatalf("test clock did not cross both recovery boundaries: created=%s recovery=%s", time.UnixMilli(operation.CreatedAtMS).UTC(), clock.Now())
|
||
}
|
||
if _, err := svc.CloseRoom(ctx, &roomv1.CloseRoomRequest{
|
||
Meta: &roomv1.RequestMeta{CommandId: "close-during-gift-recovery", ActorUserId: senderID, RoomId: roomID, AppCode: appcode.Default},
|
||
}); !xerr.IsCode(err, xerr.Conflict) {
|
||
t.Fatalf("room close must wait for debited gift recovery, got %v", err)
|
||
}
|
||
|
||
workerCtx, cancel := context.WithCancel(ctx)
|
||
defer cancel()
|
||
go svc.RunGiftOperationRecoveryWorker(workerCtx, 10*time.Millisecond)
|
||
deadline := time.Now().Add(3 * time.Second)
|
||
for {
|
||
record, committed, err := repository.GetCommand(ctx, roomID, req.GetMeta().GetCommandId())
|
||
if err != nil {
|
||
t.Fatalf("read recovered command: %v", err)
|
||
}
|
||
if committed {
|
||
if len(record.ResultPayload) == 0 {
|
||
t.Fatal("recovered command did not persist V2 result")
|
||
}
|
||
break
|
||
}
|
||
if time.Now().After(deadline) {
|
||
t.Fatal("server recovery worker did not commit debited gift")
|
||
}
|
||
time.Sleep(20 * time.Millisecond)
|
||
}
|
||
if operation.CreatedAtMS == walletPaidAtMS {
|
||
t.Fatalf("test must distinguish room saga time from wallet transaction time: %d", walletPaidAtMS)
|
||
}
|
||
if firstPaidAt, recoveredPaidAt := lucky.paidAtMS[0].Load(), lucky.paidAtMS[1].Load(); firstPaidAt != walletPaidAtMS || recoveredPaidAt != walletPaidAtMS {
|
||
t.Fatalf("lucky paid_at_ms drifted from wallet fact across recovery: wallet=%d operation=%d first=%d recovered=%d", walletPaidAtMS, operation.CreatedAtMS, firstPaidAt, recoveredPaidAt)
|
||
}
|
||
|
||
replayed, err := svc.SendGift(ctx, req)
|
||
if err != nil {
|
||
t.Fatalf("replay recovered gift failed: %v", err)
|
||
}
|
||
// HTTP 送礼回执只返回 wallet debit 的稳定账后余额；幸运返奖由异步钱包/IM 事实随后推进，
|
||
// 不能把 lucky-service 的临时 coin_balance_after 冒充同一 sender 版本的扣费余额。
|
||
if !replayed.GetV2().GetIdempotentReplay() || replayed.GetLuckyGift().GetDrawId() != "draw-recovered" || replayed.GetCoinBalanceAfter() != 900 || len(replayed.GetV2().GetLuckyHits()) != 1 || replayed.GetV2().GetLuckyHits()[0].GetGiftIndex() != 1 {
|
||
t.Fatalf("recovered original result was not replayed: %+v", replayed)
|
||
}
|
||
if _, err := svc.CloseRoom(ctx, &roomv1.CloseRoomRequest{
|
||
Meta: &roomv1.RequestMeta{CommandId: "close-after-gift-recovery", ActorUserId: senderID, RoomId: roomID, AppCode: appcode.Default},
|
||
}); err != nil {
|
||
t.Fatalf("room close should proceed after gift saga commits: %v", err)
|
||
}
|
||
}
|
||
|
||
func TestSendGiftOwnerRejectsExcessiveUnitsBeforeWallet(t *testing.T) {
|
||
ctx := appcode.WithContext(context.Background(), appcode.Default)
|
||
repository := mysqltest.NewRepository(t)
|
||
wallet := &concurrentV2Wallet{}
|
||
svc := roomservice.New(roomservice.Config{NodeID: "node-gift-v2-limit", LeaseTTL: 10 * time.Second, SnapshotEveryN: 1}, router.NewMemoryDirectory(), repository, wallet, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher())
|
||
roomID := "room-gift-v2-limit"
|
||
createRocketRoom(t, ctx, svc, roomID, 401, 9001)
|
||
targets := []int64{402, 403, 404, 405, 406, 407}
|
||
for _, targetID := range targets {
|
||
joinRocketRoom(t, ctx, svc, roomID, targetID)
|
||
}
|
||
_, err := svc.SendGift(ctx, &roomv1.SendGiftRequest{
|
||
Meta: &roomv1.RequestMeta{CommandId: "cmd-v2-limit", ActorUserId: 401, RoomId: roomID, AppCode: appcode.Default},
|
||
TargetType: "user", TargetUserIds: targets, GiftId: "rose", GiftCount: 999,
|
||
})
|
||
if !xerr.IsCode(err, xerr.InvalidArgument) {
|
||
t.Fatalf("owner total unit limit should reject request, got %v", err)
|
||
}
|
||
if wallet.calls.Load() != 0 {
|
||
t.Fatalf("rejected request reached wallet %d times", wallet.calls.Load())
|
||
}
|
||
}
|
||
|
||
var _ integration.WalletClient = (*concurrentV2Wallet)(nil)
|
||
var _ integration.LuckyGiftClient = (*recoverableLuckyClient)(nil)
|