diff --git a/services/statistics-service/internal/app/local_flow_test.go b/services/statistics-service/internal/app/local_flow_test.go index 13c53cb5..01a66db1 100644 --- a/services/statistics-service/internal/app/local_flow_test.go +++ b/services/statistics-service/internal/app/local_flow_test.go @@ -334,6 +334,33 @@ func TestLocalSelfGameStatisticsFlow(t *testing.T) { if err := repo.ConsumeSelfGamePoolAdjustment(appCtx, pool); err != nil { t.Fatalf("consume self game pool failed: %v", err) } + for _, dayOffset := range []int{1, 7} { + if err := repo.ConsumeSelfGameMatch(appCtx, mysqlstorage.SelfGameMatchEvent{ + AppCode: appCode, + EventID: "SelfGameMatchSettled:dice_retention_" + string(rune('0'+dayOffset)), + EventType: gamemq.EventTypeSelfGameMatchSettled, + MatchID: "dice_retention", + GameID: "dice", + RegionID: regionID, + StakeCoin: 500, + UserParticipants: 1, + WinnerCount: 1, + UserStakeCoin: 500, + PayoutCoin: 900, + Participants: []mysqlstorage.SelfGameMatchParticipantEvent{{ + UserID: userID, + ParticipantType: "user", + StakeCoin: 500, + DicePoints: []int64{5}, + Result: "win", + PayoutCoin: 900, + NetWinCoin: 400, + }}, + OccurredAtMS: time.UnixMilli(occurredAt).AddDate(0, 0, dayOffset).UnixMilli(), + }); err != nil { + t.Fatalf("consume retention self game match day+%d failed: %v", dayOffset, err) + } + } overview, err := repo.QuerySelfGameOverview(ctx, mysqlstorage.SelfGameOverviewQuery{ AppCode: appCode, @@ -360,6 +387,12 @@ func TestLocalSelfGameStatisticsFlow(t *testing.T) { if len(overview.UserRisk) != 1 || overview.UserRisk[0].UserID != userID || overview.UserRisk[0].WinCount != 1 || overview.UserRisk[0].NetWinCoin != 400 { t.Fatalf("self game user risk mismatch: %+v", overview.UserRisk) } + if overview.Retention.CohortUsers != 1 || overview.Retention.Day1Users != 1 || overview.Retention.Day7Users != 1 || overview.Retention.Day1Rate != 1 || overview.Retention.Day7Rate != 1 { + t.Fatalf("self game retention mismatch: %+v", overview.Retention) + } + if len(overview.MetricDefinitions) == 0 { + t.Fatalf("self game metric definitions missing") + } } func walletMessage(t *testing.T, message walletmq.WalletOutboxMessage) []byte { diff --git a/services/statistics-service/internal/storage/mysql/query.go b/services/statistics-service/internal/storage/mysql/query.go index 1475f361..dd3dd21f 100644 --- a/services/statistics-service/internal/storage/mysql/query.go +++ b/services/statistics-service/internal/storage/mysql/query.go @@ -177,6 +177,8 @@ type SelfGameOverview struct { PoolStats []SelfGamePoolStat `json:"pool_stats"` ParticipantDistribution []SelfGameParticipantStat `json:"participant_distribution"` UserRisk []SelfGameUserRiskStat `json:"user_risk"` + Retention SelfGameRetention `json:"retention"` + MetricDefinitions []MetricDefinition `json:"metric_definitions"` Funnel []SelfGameFunnelStep `json:"funnel"` Conversion SelfGameConversion `json:"conversion"` } @@ -279,6 +281,19 @@ type SelfGameUserRiskStat struct { CancelRate float64 `json:"cancel_rate"` } +type SelfGameRetention struct { + CohortUsers int64 `json:"cohort_users"` + Day1Users int64 `json:"day1_users"` + Day7Users int64 `json:"day7_users"` + Day1Rate float64 `json:"day1_rate"` + Day7Rate float64 `json:"day7_rate"` +} + +type MetricDefinition struct { + Metric string `json:"metric"` + Definition string `json:"definition"` +} + type SelfGameFunnelStep struct { EventName string `json:"event_name"` EventCount int64 `json:"event_count"` @@ -419,6 +434,12 @@ func (r *Repository) QuerySelfGameOverview(ctx context.Context, query SelfGameOv return SelfGameOverview{}, err } overview.UserRisk = risk + retention, err := r.querySelfGameRetention(ctx, app, startDay, endDay, gameID, query.CountryID, query.RegionID) + if err != nil { + return SelfGameOverview{}, err + } + overview.Retention = retention + overview.MetricDefinitions = selfGameMetricDefinitions() return overview, nil } @@ -1062,6 +1083,63 @@ func (r *Repository) querySelfGameUserRisk(ctx context.Context, app, startDay, e return out, rows.Err() } +func (r *Repository) querySelfGameRetention(ctx context.Context, app, startDay, endDay, gameID string, countryID int64, regionID int64) (SelfGameRetention, error) { + cohortFilter, cohortArgs := selfGameUserDayFilter(app, startDay, endDay, gameID, countryID, regionID) + returnFilter, returnArgs := selfGameUserDayFilter(app, addStatDays(startDay, 1), addStatDays(endDay, 7), gameID, countryID, regionID) + args := make([]any, 0, len(cohortArgs)+len(returnArgs)*2) + args = append(args, cohortArgs...) + args = append(args, returnArgs...) + args = append(args, returnArgs...) + row := r.db.QueryRowContext(ctx, ` + SELECT COUNT(1), + COALESCE(SUM(CASE WHEN d1.user_id IS NULL THEN 0 ELSE 1 END), 0), + COALESCE(SUM(CASE WHEN d7.user_id IS NULL THEN 0 ELSE 1 END), 0) + FROM ( + SELECT DISTINCT stat_day, user_id + FROM stat_self_game_user_day + WHERE `+cohortFilter+` + ) cohort + LEFT JOIN ( + SELECT DISTINCT stat_day, user_id + FROM stat_self_game_user_day + WHERE `+returnFilter+` + ) d1 ON d1.user_id = cohort.user_id AND d1.stat_day = DATE_ADD(cohort.stat_day, INTERVAL 1 DAY) + LEFT JOIN ( + SELECT DISTINCT stat_day, user_id + FROM stat_self_game_user_day + WHERE `+returnFilter+` + ) d7 ON d7.user_id = cohort.user_id AND d7.stat_day = DATE_ADD(cohort.stat_day, INTERVAL 7 DAY)`, args...) + var retention SelfGameRetention + if err := row.Scan(&retention.CohortUsers, &retention.Day1Users, &retention.Day7Users); err != nil { + return SelfGameRetention{}, err + } + retention.Day1Rate = ratio(retention.Day1Users, retention.CohortUsers) + retention.Day7Rate = ratio(retention.Day7Users, retention.CohortUsers) + return retention, nil +} + +func selfGameUserDayFilter(app, startDay, endDay, gameID string, countryID int64, regionID int64) (string, []any) { + filter := "app_code = ? AND match_count > 0" + args := []any{app} + if startDay != "" && endDay != "" { + filter += " AND stat_day BETWEEN ? AND ?" + args = append(args, startDay, endDay) + } + if gameID != "" { + filter += " AND game_id = ?" + args = append(args, gameID) + } + if countryID > 0 { + filter += " AND country_id = ?" + args = append(args, countryID) + } + if regionID > 0 { + filter += " AND region_id = ?" + args = append(args, regionID) + } + return filter, args +} + func selfGameFilter(app, startDay, endDay, gameID string, countryID int64, regionID int64) (string, []any) { filter := "app_code = ? AND stat_day BETWEEN ? AND ?" args := []any{app, startDay, endDay} @@ -1146,6 +1224,14 @@ func previousStatDayRange(startDay, endDay string) (string, string) { return previousStart.Format("2006-01-02"), previousEnd.Format("2006-01-02") } +func addStatDays(day string, offset int) string { + parsed, err := time.Parse("2006-01-02", day) + if err != nil { + return day + } + return parsed.AddDate(0, 0, offset).Format("2006-01-02") +} + func div(numerator, denominator int64) int64 { if denominator <= 0 { return 0 @@ -1172,3 +1258,18 @@ func ratio(numerator, denominator int64) float64 { } return float64(numerator) / float64(denominator) } + +func selfGameMetricDefinitions() []MetricDefinition { + return []MetricDefinition{ + {Metric: "game_pv", Definition: "H5 page_open event_count；同一用户多次打开会重复计数。"}, + {Metric: "game_uv", Definition: "H5 page_open user_count；同一 UTC 日、同一 game_id、同一用户只计一次。"}, + {Metric: "match_click_users", Definition: "H5 match_click user_count；用户点击匹配按钮后上报。"}, + {Metric: "match_success_users", Definition: "H5 match_success user_count；匹配接口成功返回有效 match_id 后上报。"}, + {Metric: "settlement_users", Definition: "H5 settlement_show user_count；结算页展示时上报。"}, + {Metric: "retention_cohort_users", Definition: "统计窗口内有已结算自研游戏对局的真实用户；只统计 stat_self_game_user_day.match_count > 0。"}, + {Metric: "day1_retention", Definition: "cohort 用户在第 1 个 UTC 统计日再次有已结算自研游戏对局；game_id、country_id、region_id 筛选会同时作用于 cohort 和回访日。"}, + {Metric: "day7_retention", Definition: "cohort 用户在第 7 个 UTC 统计日再次有已结算自研游戏对局；game_id、country_id、region_id 筛选会同时作用于 cohort 和回访日。"}, + {Metric: "platform_profit_coin", Definition: "结算事件快照的 user_stake_coin + robot_stake_coin - payout_coin - refund_coin；与奖池入池/出池流水分表展示。"}, + {Metric: "pool_stats", Definition: "game-service 奖池调整流水聚合；in_coin/out_coin 是入池/出池金额，balance_after_coin 是当天同方向同原因聚合行内最新余额快照。"}, + } +} diff --git a/services/user-service/internal/storage/mysql/cp/repository.go b/services/user-service/internal/storage/mysql/cp/repository.go index 17469a51..0a17beb0 100644 --- a/services/user-service/internal/storage/mysql/cp/repository.go +++ b/services/user-service/internal/storage/mysql/cp/repository.go @@ -10,6 +10,7 @@ import ( "time" "hyapp/pkg/appcode" + "hyapp/pkg/idgen" "hyapp/pkg/xerr" cpdomain "hyapp/services/user-service/internal/domain/cp" ) @@ -1144,7 +1145,8 @@ func insertOutboxTx(ctx context.Context, tx *sql.Tx, appCode string, eventType s if err != nil { return err } - eventID := fmt.Sprintf("%s:%s:%d:%d", aggregateType, eventType, aggregateID, nowMs) + // event_id 是 user_outbox 的投递事实主键，必须按每条通知唯一生成；同一个目标用户同毫秒内可能收到多条 CP 申请/结果通知。 + eventID := idgen.New("uout") // outbox 与业务状态同事务提交；RocketMQ 发布失败只影响异步 IM/统计，不回滚已经同意的关系。 _, err = tx.ExecContext(ctx, ` INSERT INTO user_outbox ( diff --git a/services/user-service/internal/storage/mysql/cp/repository_test.go b/services/user-service/internal/storage/mysql/cp/repository_test.go index 5b04dfa9..6f83eba2 100644 --- a/services/user-service/internal/storage/mysql/cp/repository_test.go +++ b/services/user-service/internal/storage/mysql/cp/repository_test.go @@ -271,6 +271,169 @@ func TestConsumeGiftEventIgnoresNormalGiftWithoutRelationship(t *testing.T) { } } +func TestConsumeGiftEventCreatesNewApplicationAfterRejected(t *testing.T) { + schema := mysqlschema.New(t) + repo := New(schema.DB) + ctx := appcode.WithContext(context.Background(), appcode.Default) + nowMS := int64(1700000000000) + + seedCPTestUser(t, schema.DB, 4401, "Invite Alice", nowMS) + seedCPTestUser(t, schema.DB, 4402, "Invite Bob", nowMS) + + // 第一次 CP 礼物只生成 pending 申请；目标用户拒绝后，这条申请必须保持终态，不允许后续礼物回写成 pending。 + first, err := repo.ConsumeGiftEvent(ctx, cpdomain.GiftEvent{ + AppCode: appcode.Default, + EventID: "room_evt_cp_invite_1", + RoomID: "room-cp", + RoomVersion: 3, + OccurredAtMS: nowMS + 1, + SenderUserID: 4401, + TargetUserID: 4402, + GiftID: "cp_gift", + GiftCount: 1, + GiftValue: 37, + BillingReceiptID: "receipt-cp-invite-1", + CommandID: "cmd-cp-invite-1", + CPRelationType: cpdomain.RelationTypeCP, + GiftName: "CP Gift", + }, nowMS+2) + if err != nil { + t.Fatalf("consume first cp gift failed: %v", err) + } + if !first.Consumed || first.Application.Status != cpdomain.ApplicationStatusPending { + t.Fatalf("first cp gift must create pending application: %+v", first) + } + if _, err := repo.RejectApplication(ctx, 4402, first.Application.ApplicationID, "declined", nowMS+3); err != nil { + t.Fatalf("reject first cp application failed: %v", err) + } + + // 第二次送礼来自新的 room outbox event，应创建新的 pending 申请，不能复用或覆盖旧 rejected 申请。 + second, err := repo.ConsumeGiftEvent(ctx, cpdomain.GiftEvent{ + AppCode: appcode.Default, + EventID: "room_evt_cp_invite_2", + RoomID: "room-cp", + RoomVersion: 4, + OccurredAtMS: nowMS + 4, + SenderUserID: 4401, + TargetUserID: 4402, + GiftID: "cp_gift", + GiftCount: 1, + GiftValue: 37, + BillingReceiptID: "receipt-cp-invite-2", + CommandID: "cmd-cp-invite-2", + CPRelationType: cpdomain.RelationTypeCP, + GiftName: "CP Gift", + }, nowMS+5) + if err != nil { + t.Fatalf("consume second cp gift failed: %v", err) + } + if !second.Consumed || second.Application.Status != cpdomain.ApplicationStatusPending { + t.Fatalf("second cp gift must create pending application: %+v", second) + } + if second.Application.ApplicationID == first.Application.ApplicationID { + t.Fatalf("rejected application must not be reused: first=%s second=%s", first.Application.ApplicationID, second.Application.ApplicationID) + } + + statuses := map[string]string{} + rows, err := schema.DB.QueryContext(ctx, ` + SELECT application_id, status + FROM user_cp_applications + WHERE app_code = ? AND requester_user_id = ? AND target_user_id = ? + ORDER BY created_at_ms ASC`, + appcode.Default, 4401, 4402, + ) + if err != nil { + t.Fatalf("list cp applications failed: %v", err) + } + defer rows.Close() + for rows.Next() { + var applicationID string + var status string + if err := rows.Scan(&applicationID, &status); err != nil { + t.Fatalf("scan cp application status failed: %v", err) + } + statuses[applicationID] = status + } + if err := rows.Err(); err != nil { + t.Fatalf("iterate cp application statuses failed: %v", err) + } + if statuses[first.Application.ApplicationID] != cpdomain.ApplicationStatusRejected { + t.Fatalf("first application must remain rejected: %+v", statuses) + } + if statuses[second.Application.ApplicationID] != cpdomain.ApplicationStatusPending { + t.Fatalf("second application must remain pending: %+v", statuses) + } +} + +func TestConsumeGiftEventWritesDistinctOutboxEventsInSameMillisecond(t *testing.T) { + schema := mysqlschema.New(t) + repo := New(schema.DB) + ctx := appcode.WithContext(context.Background(), appcode.Default) + nowMS := int64(1700000000000) + + seedCPTestUser(t, schema.DB, 4501, "Outbox Alice", nowMS) + seedCPTestUser(t, schema.DB, 4502, "Outbox Bob", nowMS) + + // 同一 pending 申请被后续礼物刷新时仍会产生新的通知事实；event_id 不能只按目标用户和毫秒拼接，否则 IM 会被主键碰撞吞掉。 + first, err := repo.ConsumeGiftEvent(ctx, cpdomain.GiftEvent{ + AppCode: appcode.Default, + EventID: "room_evt_cp_outbox_1", + RoomID: "room-cp-outbox", + RoomVersion: 3, + OccurredAtMS: nowMS + 1, + SenderUserID: 4501, + TargetUserID: 4502, + GiftID: "cp_gift", + GiftCount: 1, + GiftValue: 37, + BillingReceiptID: "receipt-cp-outbox-1", + CommandID: "cmd-cp-outbox-1", + CPRelationType: cpdomain.RelationTypeCP, + GiftName: "CP Gift", + }, nowMS+2) + if err != nil { + t.Fatalf("consume first cp outbox gift failed: %v", err) + } + second, err := repo.ConsumeGiftEvent(ctx, cpdomain.GiftEvent{ + AppCode: appcode.Default, + EventID: "room_evt_cp_outbox_2", + RoomID: "room-cp-outbox", + RoomVersion: 4, + OccurredAtMS: nowMS + 2, + SenderUserID: 4501, + TargetUserID: 4502, + GiftID: "cp_gift", + GiftCount: 1, + GiftValue: 37, + BillingReceiptID: "receipt-cp-outbox-2", + CommandID: "cmd-cp-outbox-2", + CPRelationType: cpdomain.RelationTypeCP, + GiftName: "CP Gift", + }, nowMS+2) + if err != nil { + t.Fatalf("consume second cp outbox gift failed: %v", err) + } + if first.Application.ApplicationID != second.Application.ApplicationID { + t.Fatalf("same pending application should be refreshed before a decision: first=%s second=%s", first.Application.ApplicationID, second.Application.ApplicationID) + } + + var outboxCount int64 + if err := schema.DB.QueryRowContext(ctx, ` + SELECT COUNT(*) + FROM user_outbox + WHERE app_code = ? + AND event_type = ? + AND aggregate_type = 'cp_application' + AND aggregate_id = ?`, + appcode.Default, cpdomain.EventTypeApplicationCreated, 4502, + ).Scan(&outboxCount); err != nil { + t.Fatalf("count cp outbox events failed: %v", err) + } + if outboxCount != 2 { + t.Fatalf("same-millisecond cp notifications must keep two outbox rows, got %d", outboxCount) + } +} + func TestBreakRelationshipPrepareConfirmIsIdempotent(t *testing.T) { schema := mysqlschema.New(t) repo := New(schema.DB)