diff --git a/internal/data/business/channel_metrics.go b/internal/data/business/channel_metrics.go index 95863de..d2e6875 100644 --- a/internal/data/business/channel_metrics.go +++ b/internal/data/business/channel_metrics.go @@ -2,11 +2,37 @@ package business import ( "context" + "sort" "gorm.io/gorm" "kra/internal/biz/business" ) +func whereMetricChannels(db *gorm.DB, keys []business.ChannelKey) *gorm.DB { + bySource := map[string][]uint64{} + for _, key := range keys { + bySource[key.SourceDB] = append(bySource[key.SourceDB], key.ID) + } + sources := make([]string, 0, len(bySource)) + for source := range bySource { + sources = append(sources, source) + } + sort.Strings(sources) + if len(sources) == 0 { + return db.Where("1=0") + } + var condition *gorm.DB + for _, source := range sources { + part := db.Session(&gorm.Session{NewDB: true}).Where("source_db=? AND promote_id IN ?", source, bySource[source]) + if condition == nil { + condition = part + } else { + condition = condition.Or(part) + } + } + return db.Where(condition) +} + func (r *channelRepo) Metrics(ctx context.Context, roots, statKeys []business.ChannelKey, dedup bool) (*business.ChannelMetricData, error) { db := r.data.DB().WithContext(ctx) metrics := &business.ChannelMetricData{Games: map[business.ChannelKey]int64{}, Players: map[business.ChannelKey]int64{}, Pay: map[business.ChannelKey]string{}} @@ -31,7 +57,7 @@ func (r *channelRepo) Metrics(ctx context.Context, roots, statKeys []business.Ch if metrics.GamesAvailable { for start := 0; start < len(roots); start += 300 { var rows []countRow - if err := db.Table("tab_promote_apply").Select("source_db,promote_id,COUNT(DISTINCT game_id) AS count").Where("status = 1 AND (source_db,promote_id) IN ?", channelPairs(roots[start:min(start+300, len(roots))])).Group("source_db,promote_id").Scan(&rows).Error; err != nil { + if err := whereMetricChannels(db, roots[start:min(start+300, len(roots))]).Table("tab_promote_apply").Select("source_db,promote_id,COUNT(DISTINCT game_id) AS count").Where("status = 1").Group("source_db,promote_id").Scan(&rows).Error; err != nil { return nil, business.ErrStorage } for _, row := range rows { @@ -41,9 +67,9 @@ func (r *channelRepo) Metrics(ctx context.Context, roots, statKeys []business.Ch } // 每批按平台/渠道一起查询,避免逐个渠道访问数据库,也避免参数数量无限增长。 for start := 0; start < len(statKeys); start += 300 { - pairs := channelPairs(statKeys[start:min(start+300, len(statKeys))]) + keys := statKeys[start:min(start+300, len(statKeys))] if metrics.PlayersAvailable { - q := db.Table("tab_user").Where("(puid = 0 OR puid IS NULL) AND (source_db,promote_id) IN ?", pairs) + q := whereMetricChannels(db, keys).Table("tab_user").Where("puid = 0 OR puid IS NULL") if dedup { var players []struct { SourceDB string `gorm:"column:source_db"` @@ -72,7 +98,7 @@ func (r *channelRepo) Metrics(ctx context.Context, roots, statKeys []business.Ch PromoteID uint64 Amount string } - if err := db.Table("tab_spend").Select("source_db,promote_id,COALESCE(SUM(pay_amount),0) AS amount").Where("pay_status = 1 AND (source_db,promote_id) IN ?", pairs).Group("source_db,promote_id").Scan(&rows).Error; err != nil { + if err := whereMetricChannels(db, keys).Table("tab_spend").Select("source_db,promote_id,COALESCE(SUM(pay_amount),0) AS amount").Where("pay_status = 1").Group("source_db,promote_id").Scan(&rows).Error; err != nil { return nil, business.ErrStorage } for _, row := range rows { diff --git a/internal/data/business/channel_sort.go b/internal/data/business/channel_sort.go index c071a8b..9ec5a47 100644 --- a/internal/data/business/channel_sort.go +++ b/internal/data/business/channel_sort.go @@ -54,6 +54,10 @@ func channelMetricPage(db, roots *gorm.DB, f business.ChannelFilter, o business. index := "idx_sw_user_report_identity" if table == "tab_spend" { index = "idx_sw_spend_promote_summary" + } else if !f.IDCardDedup { + // Its source/channel prefix supports both filtering and grouped counting, + // without sorting the much wider identity index into a temporary table. + index = "idx_alert_user_window" } if db.Dialector.Name() == "mysql" && db.Migrator().HasIndex(table, index) { metricTable += " FORCE INDEX (" + index + ")" @@ -112,14 +116,20 @@ func channelMetricPage(db, roots *gorm.DB, f business.ChannelFilter, o business. Score string } if f.SortField == "totalPayAmount" { - leaves = leaves.Select("source_db,promote_id,COALESCE(SUM(pay_amount),0) AS score") + leaves = leaves.Select("source_db,promote_id AS root_id,promote_id,COALESCE(SUM(pay_amount),0) AS score") } else { - leaves = leaves.Select("source_db,promote_id,COUNT(*) AS score") + leaves = leaves.Select("source_db,promote_id AS root_id,promote_id,COUNT(*) AS score") } leaves = leaves.Group("source_db,promote_id") - q = q.Joins("JOIN (?) leaf ON leaf.source_db=cm.source_db AND leaf.promote_id=cm.leaf_id", leaves). - Select("cm.source_db,cm.root_id,SUM(leaf.score) AS score") - if err := q.Group("cm.source_db,cm.root_id").Scan(&values).Error; err != nil { + if f.IncludeSub { + q = q.Joins("JOIN (?) leaf ON leaf.source_db=cm.source_db AND leaf.promote_id=cm.leaf_id", leaves). + Select("cm.source_db,cm.root_id,SUM(leaf.score) AS score").Group("cm.source_db,cm.root_id") + } else { + // A channel's own score needs no membership join or second grouping. + // Scores remain internal; pagination below only returns visible roots. + q = leaves + } + if err := q.Scan(&values).Error; err != nil { return nil, business.ErrStorage } for _, value := range values { diff --git a/internal/data/business/mysql_rehearsal_test.go b/internal/data/business/mysql_rehearsal_test.go index de9ea9f..8c0408e 100644 --- a/internal/data/business/mysql_rehearsal_test.go +++ b/internal/data/business/mysql_rehearsal_test.go @@ -262,6 +262,37 @@ func TestMySQLLaterRegistrationIndexesPreserveMalformedLegacyTimes(t *testing.T) } } +func TestMySQLChannelOwnPlayerSortUsesGroupedCoveringIndex(t *testing.T) { + db := openRehearsalMySQL(t) + for _, sql := range []string{ + "CREATE TABLE tab_promote(id BIGINT,source_db VARCHAR(10),create_time BIGINT,PRIMARY KEY(id,source_db))", + "CREATE TABLE tab_user(id BIGINT,source_db VARCHAR(10),promote_id BIGINT,puid BIGINT,sw_register_time_unix BIGINT,idcard VARCHAR(20),PRIMARY KEY(id,source_db),INDEX idx_alert_user_window(source_db,promote_id,sw_register_time_unix,puid))", + "INSERT INTO tab_promote VALUES(1,'lx',1),(2,'lx',2),(1,'hw',3)", + "INSERT INTO tab_user VALUES(1,'lx',1,0,1,'a'),(2,'lx',1,0,1,'a'),(3,'lx',2,0,1,'b'),(1,'hw',1,0,1,'c')", + } { + if err := db.Exec(sql).Error; err != nil { + t.Fatal(err) + } + } + keys, err := channelMetricPage(db, db.Table("tab_promote p").Where("p.source_db='lx'"), business.ChannelFilter{Sources: []string{"lx"}, SortField: "playerCount"}, business.NewListOptions(business.ListLimit(1))) + if err != nil || len(keys) != 1 || keys[0] != (business.ChannelKey{SourceDB: "lx", ID: 1}) { + t.Fatalf("own-channel sort: %v %v", keys, err) + } + var counts []struct { + SourceDB string + PromoteID uint64 + Count int64 + } + if err := whereMetricChannels(db, []business.ChannelKey{{SourceDB: "lx", ID: 2}, {SourceDB: "hw", ID: 1}}).Table("tab_user").Select("source_db,promote_id,COUNT(*) AS count").Group("source_db,promote_id").Scan(&counts).Error; err != nil || len(counts) != 2 { + t.Fatalf("paired source filter: %v %v", counts, err) + } + for _, row := range counts { + if row.Count != 1 { + t.Fatalf("included a same-id channel from another platform: %+v", row) + } + } +} + func TestMySQLChannelCreationSerializesAccount(t *testing.T) { db := openRehearsalMySQL(t) if err := db.AutoMigrate(&platformChannelPO{}, &channelSettlementPO{}); err != nil {