幂等
This commit is contained in:
zhx
parent
8628de3dbd
commit
88551cd96d
@ -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 {
|
||||
|
||||
@ -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 是当天同方向同原因聚合行内最新余额快照。"},
|
||||
}
|
||||
}
|
||||
|
||||
@ -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 (
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user