diff --git a/services/activity-service/internal/domain/task/task.go b/services/activity-service/internal/domain/task/task.go index 87146c20..ada8175b 100644 --- a/services/activity-service/internal/domain/task/task.go +++ b/services/activity-service/internal/domain/task/task.go @@ -13,13 +13,20 @@ const ( AudienceNewbie = "newbie" ActionNone = "none" + ActionRoom = "room" ActionRoomRandom = "room_random" + ActionRoomWindow = "room_window" + ActionRoomRandomWindow = "room_random_window" + ActionRoomGame = "room_game" ActionRoomRandomGame = "room_random_game" + ActionExploreGame = "explore_game" ActionWallet = "wallet" ActionGiftPanel = "gift_panel" ActionGiftPanelSpecific = "gift_panel_specific" MetricGameSpendCoin = "game_spend_coin" + MetricGameDicePlayCount = "game_dice_play_count" + MetricGameRockPlayCount = "game_rock_play_count" MetricGiftSpendCoin = "gift_spend_coin" MetricGiftSendCount = "gift_send_count" MetricLuckyGiftSpendCoin = "lucky_gift_spend_coin" diff --git a/services/activity-service/internal/service/task/service.go b/services/activity-service/internal/service/task/service.go index e50f2be0..7a2f223c 100644 --- a/services/activity-service/internal/service/task/service.go +++ b/services/activity-service/internal/service/task/service.go @@ -535,10 +535,20 @@ func normalizeActionType(value string) string { switch strings.ToLower(strings.TrimSpace(value)) { case "", taskdomain.ActionNone: return taskdomain.ActionNone + case taskdomain.ActionRoom: + return taskdomain.ActionRoom case taskdomain.ActionRoomRandom: return taskdomain.ActionRoomRandom + case taskdomain.ActionRoomWindow: + return taskdomain.ActionRoomWindow + case taskdomain.ActionRoomRandomWindow: + return taskdomain.ActionRoomRandomWindow + case taskdomain.ActionRoomGame: + return taskdomain.ActionRoomGame case taskdomain.ActionRoomRandomGame: return taskdomain.ActionRoomRandomGame + case taskdomain.ActionExploreGame: + return taskdomain.ActionExploreGame case taskdomain.ActionWallet: return taskdomain.ActionWallet case taskdomain.ActionGiftPanel: @@ -552,7 +562,17 @@ func normalizeActionType(value string) string { func validActionType(value string) bool { switch value { - case taskdomain.ActionNone, taskdomain.ActionRoomRandom, taskdomain.ActionRoomRandomGame, taskdomain.ActionWallet, taskdomain.ActionGiftPanel, taskdomain.ActionGiftPanelSpecific: + case taskdomain.ActionNone, + taskdomain.ActionRoom, + taskdomain.ActionRoomRandom, + taskdomain.ActionRoomWindow, + taskdomain.ActionRoomRandomWindow, + taskdomain.ActionRoomGame, + taskdomain.ActionRoomRandomGame, + taskdomain.ActionExploreGame, + taskdomain.ActionWallet, + taskdomain.ActionGiftPanel, + taskdomain.ActionGiftPanelSpecific: return true default: return false diff --git a/services/game-service/internal/service/game/service.go b/services/game-service/internal/service/game/service.go index 1f92ef4a..8f243f5c 100644 --- a/services/game-service/internal/service/game/service.go +++ b/services/game-service/internal/service/game/service.go @@ -26,6 +26,13 @@ import ( const defaultScene = gamedomain.SceneVoiceRoom +const ( + selfGameIDDice = "dice" + selfGameIDRock = "rock" + taskMetricGameDicePlayed = "game_dice_play_count" + taskMetricGameRockPlayed = "game_rock_play_count" +) + // Repository 隔离 game-service 持久化事实,订单和 session 不能只放内存。 type Repository interface { Ping(ctx context.Context) error @@ -697,7 +704,38 @@ func (s *Service) ProcessLevelEventOutboxBatch(ctx context.Context, runID string _ = s.repository.MarkLevelEventFailed(eventCtx, event.EventID, "task returned "+taskResp.GetStatus(), nextRetry, s.now().UnixMilli()) continue } - // 只有成长和任务两条链路都达到终态才标记 delivered,防止“游戏等级已加、任务未加”的半完成 outbox。 + if playMetric := selfGamePlayCountMetric(event.GameID); playMetric != "" { + // dice/rock 的“玩 1 局”是独立每日任务口径,不能复用扣币金额;仍使用同一条扣费 outbox 派生,保证每个真人扣款参与者最多累计 1 次。 + playResp, relayErr := s.activity.ConsumeTaskEvent(eventCtx, &activityv1.ConsumeTaskEventRequest{ + Meta: &activityv1.RequestMeta{ + RequestId: event.EventID + ":task:play_count", + Caller: "game-service", + SentAtMs: s.now().UnixMilli(), + AppCode: event.AppCode, + }, + EventId: event.EventID + ":task:play_count", + EventType: "SelfGamePlayed", + SourceService: "game-service", + UserId: event.UserID, + MetricType: playMetric, + Value: 1, + OccurredAtMs: event.OccurredAtMS, + DimensionsJson: event.PayloadJSON, + }) + if relayErr != nil { + failure++ + nextRetry := s.now().Add(time.Minute).UnixMilli() + _ = s.repository.MarkLevelEventFailed(eventCtx, event.EventID, xerr.MessageOf(relayErr), nextRetry, s.now().UnixMilli()) + continue + } + if playResp.GetStatus() != "consumed" && playResp.GetStatus() != "skipped" && playResp.GetStatus() != "duplicate" { + failure++ + nextRetry := s.now().Add(time.Minute).UnixMilli() + _ = s.repository.MarkLevelEventFailed(eventCtx, event.EventID, "play task returned "+playResp.GetStatus(), nextRetry, s.now().UnixMilli()) + continue + } + } + // 只有成长、扣币任务和自研游戏局数任务都达到终态才标记 delivered,防止“游戏等级已加、每日任务未加”的半完成 outbox。 if err := s.repository.MarkLevelEventDelivered(eventCtx, event.EventID, s.now().UnixMilli()); err != nil { failure++ continue @@ -707,6 +745,17 @@ func (s *Service) ProcessLevelEventOutboxBatch(ctx context.Context, runID string return int32(len(events)), processed, success, failure, len(events) == batchSize, nil } +func selfGamePlayCountMetric(gameID string) string { + switch strings.TrimSpace(gameID) { + case selfGameIDDice: + return taskMetricGameDicePlayed + case selfGameIDRock: + return taskMetricGameRockPlayed + default: + return "" + } +} + func normalizeScene(scene string) string { scene = strings.ToLower(strings.TrimSpace(scene)) if scene == "" { diff --git a/services/game-service/internal/service/game/service_test.go b/services/game-service/internal/service/game/service_test.go index 8b2a5db0..b4df1926 100644 --- a/services/game-service/internal/service/game/service_test.go +++ b/services/game-service/internal/service/game/service_test.go @@ -1653,6 +1653,42 @@ func TestProcessLevelEventOutboxBatchRelaysGameSpend(t *testing.T) { } } +func TestProcessLevelEventOutboxBatchRelaysSelfGamePlayCount(t *testing.T) { + repo := &fakeRepository{levelEvents: []gamedomain.LevelEventOutbox{{ + AppCode: "lalu", + EventID: "game_level:dice_order_1", + OrderID: "dice_order_1", + UserID: 42, + GameID: "dice", + ProviderOrderID: "dice_match_1:42:debit", + ProviderRoundID: "dice_match_1", + CoinAmount: 100, + WalletTransactionID: "wtx-dice-1", + PayloadJSON: `{"game_id":"dice","match_id":"dice_match_1"}`, + OccurredAtMS: 1700000000000, + }}} + activity := &fakeActivity{status: "consumed"} + svc := New(Config{}, repo, &fakeWallet{}, &fakeUser{}, activity) + svc.now = func() time.Time { return time.UnixMilli(1700000001000) } + + _, _, success, failure, _, err := svc.ProcessLevelEventOutboxBatch(context.Background(), "run_1", "worker_1", 100, 30*time.Second) + if err != nil { + t.Fatalf("ProcessLevelEventOutboxBatch failed: %v", err) + } + if success != 1 || failure != 0 { + t.Fatalf("batch result mismatch: success=%d failure=%d", success, failure) + } + if len(activity.taskRequests) != 2 { + t.Fatalf("task request count mismatch: %d", len(activity.taskRequests)) + } + if activity.taskRequests[0].GetMetricType() != "game_spend_coin" || activity.taskRequests[0].GetValue() != 100 { + t.Fatalf("game spend task mismatch: %+v", activity.taskRequests[0]) + } + if activity.taskRequests[1].GetMetricType() != "game_dice_play_count" || activity.taskRequests[1].GetValue() != 1 || activity.taskRequests[1].GetEventId() != "game_level:dice_order_1:task:play_count" { + t.Fatalf("play count task mismatch: %+v", activity.taskRequests[1]) + } +} + type fakeRepository struct { launchable gamedomain.LaunchableGame platform gamedomain.Platform @@ -1846,10 +1882,11 @@ func (f *fakeWallet) GetBalances(context.Context, *walletv1.GetBalancesRequest) } type fakeActivity struct { - status string - taskStatus string - last *activityv1.ConsumeLevelEventRequest - lastTask *activityv1.ConsumeTaskEventRequest + status string + taskStatus string + last *activityv1.ConsumeLevelEventRequest + lastTask *activityv1.ConsumeTaskEventRequest + taskRequests []*activityv1.ConsumeTaskEventRequest } func (f *fakeActivity) ConsumeLevelEvent(_ context.Context, req *activityv1.ConsumeLevelEventRequest, _ ...grpc.CallOption) (*activityv1.ConsumeLevelEventResponse, error) { @@ -1863,6 +1900,7 @@ func (f *fakeActivity) ConsumeLevelEvent(_ context.Context, req *activityv1.Cons func (f *fakeActivity) ConsumeTaskEvent(_ context.Context, req *activityv1.ConsumeTaskEventRequest, _ ...grpc.CallOption) (*activityv1.ConsumeTaskEventResponse, error) { f.lastTask = req + f.taskRequests = append(f.taskRequests, req) status := f.taskStatus if status == "" { status = "consumed"