fix: 修复商务汇总任务超时并验证计划启用

internal/data/business/migrations.go

@@ -6,7 +6,7 @@ import (
 )
 
 func Migrations() []migration.Step {
-	return []migration.Step{{ID: "202611040013_business_query_indexes", Migrate: migrateBusinessQueryIndexes}, {ID: "202611040012_business_server_report_index", Migrate: migrateServerReportIndex}, {ID: "202611040011_business_registration_index", Migrate: migrateRegistrationIndex}, {ID: "202611040010_business_alert_trigger_locks", Migrate: func(db *gorm.DB) error { return migration.CreateMissingTables(db, &alertTriggerLockPO{}) }}, {ID: "202611040009_business_menu_icons", Migrate: migrateBusinessMenuIcons}, {ID: "202611040008_business_report_build_locks", Migrate: func(db *gorm.DB) error { return migration.CreateMissingTables(db, &reportBuildLockPO{}) }}, {ID: "202611030007_business_alert_attachments", Migrate: migrateAlertAttachments}, {ID: "202611030006_business_alert_schema", Migrate: migrateAlertSchema}, {ID: "202611030005_business_password_reset", Migrate: func(db *gorm.DB) error {
+	return []migration.Step{{ID: "202611040014_business_report_build_indexes", Migrate: migrateReportBuildQueryIndexes}, {ID: "202611040013_business_query_indexes", Migrate: migrateBusinessQueryIndexes}, {ID: "202611040012_business_server_report_index", Migrate: migrateServerReportIndex}, {ID: "202611040011_business_registration_index", Migrate: migrateRegistrationIndex}, {ID: "202611040010_business_alert_trigger_locks", Migrate: func(db *gorm.DB) error { return migration.CreateMissingTables(db, &alertTriggerLockPO{}) }}, {ID: "202611040009_business_menu_icons", Migrate: migrateBusinessMenuIcons}, {ID: "202611040008_business_report_build_locks", Migrate: func(db *gorm.DB) error { return migration.CreateMissingTables(db, &reportBuildLockPO{}) }}, {ID: "202611030007_business_alert_attachments", Migrate: migrateAlertAttachments}, {ID: "202611030006_business_alert_schema", Migrate: migrateAlertSchema}, {ID: "202611030005_business_password_reset", Migrate: func(db *gorm.DB) error {
 		return migration.CreateMissingTables(db, &resetCodePO{}, &resetTokenPO{})
 	}}, {ID: "202611030004_business_binding_locks", Migrate: func(db *gorm.DB) error {
 		return migration.CreateMissingTables(db, &bindingLockPO{})

internal/data/business/query_indexes.go

@@ -22,15 +22,28 @@ var businessQueryIndexes = []businessQueryIndex{
 	{"tab_sw_report_server_day_key", "idx_sw_server_key_game_scan", []string{"game_name", "dedup_mode", "stat_date", "source_db", "game_id", "server_num", "metric", "identity_key"}},
 }
 
+var reportBuildQueryIndexes = []businessQueryIndex{
+	{"tab_user", "idx_sw_user_report_identity", []string{"id", "source_db", "puid", "promote_id", "idcard"}},
+	{"tab_user_play_info", "idx_sw_play_report_active", []string{"play_time", "user_id", "source_db", "promote_id", "game_id", "server_id"}},
+}
+
 // 只为已经接入的业务表增加索引，不创建平台表、不改写记录。
 // 按实际筛选前缀排列，并覆盖聚合读取列，避免大范围回表。
 func migrateBusinessQueryIndexes(db *gorm.DB) error {
+	return migrateQueryIndexes(db, businessQueryIndexes)
+}
+
+func migrateReportBuildQueryIndexes(db *gorm.DB) error {
+	return migrateQueryIndexes(db, reportBuildQueryIndexes)
+}
+
+func migrateQueryIndexes(db *gorm.DB, indexes []businessQueryIndex) error {
 	if db.Dialector.Name() != "mysql" {
 		return nil
 	}
 	groups := make(map[string][]businessQueryIndex)
 	tables := []string{}
-	for _, index := range businessQueryIndexes {
+	for _, index := range indexes {
 		if _, exists := groups[index.table]; !exists {
 			tables = append(tables, index.table)
 		}

internal/data/business/query_indexes_test.go

@@ -3,6 +3,8 @@ package business
 import (
 	"strings"
 	"testing"
+
+	"kra/internal/biz/business"
 )
 
 func TestMySQLBusinessQueryIndexesAreAdditiveAndIdempotent(t *testing.T) {
@@ -48,3 +50,69 @@ func TestMySQLBusinessQueryIndexesAreAdditiveAndIdempotent(t *testing.T) {
 		t.Fatalf("must reject a conflicting existing index: %v", err)
 	}
 }
+
+func TestMySQLReportBuildIndexesCoverInputsAndPreserveIdentities(t *testing.T) {
+	db := openRehearsalMySQL(t)
+	for _, sql := range []string{
+		"CREATE TABLE tab_user(id INT,source_db VARCHAR(10),puid INT,promote_id INT,idcard VARCHAR(20),payload VARCHAR(1000),PRIMARY KEY(id,source_db)) ENGINE=InnoDB",
+		"CREATE TABLE tab_user_play_info(id INT PRIMARY KEY,play_time INT,user_id INT,source_db VARCHAR(10),promote_id INT,game_id INT,server_id VARCHAR(30),payload VARCHAR(1000),KEY(play_time)) ENGINE=InnoDB",
+	} {
+		if err := db.Exec(sql).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	users, plays := []map[string]any{}, []map[string]any{}
+	known := map[business.ReportUserKey]bool{}
+	ids := []int64{}
+	for id := int64(1); id <= 4000; id++ {
+		for _, source := range []string{"lx", "hw"} {
+			users = append(users, map[string]any{"id": id, "source_db": source, "puid": 0, "promote_id": id, "idcard": source, "payload": strings.Repeat("x", 300)})
+		}
+		plays = append(plays, map[string]any{"id": id, "play_time": id, "user_id": id, "source_db": "lx", "promote_id": id, "game_id": 1, "server_id": " 001 ", "payload": strings.Repeat("x", 300)})
+		if id <= 501 {
+			known[business.ReportUserKey{SourceDB: "lx", ID: id}] = true
+			ids = append(ids, id)
+		}
+	}
+	for table, rows := range map[string][]map[string]any{"tab_user": users, "tab_user_play_info": plays} {
+		if err := db.Table(table).CreateInBatches(rows, 200).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	for attempt := 0; attempt < 2; attempt++ {
+		if err := migrateReportBuildQueryIndexes(db); err != nil {
+			t.Fatal(err)
+		}
+	}
+	loaded, err := loadReportUsers(db, known, map[int64]bool{501: true})
+	if err != nil || len(loaded) != 502 {
+		t.Fatalf("分批读取或跨平台歧义匹配错误: rows=%d error=%v", len(loaded), err)
+	}
+	for key, user := range loaded {
+		if user.IDCard != key.SourceDB || user.PromoteID == nil || *user.PromoteID != key.ID {
+			t.Fatalf("迁移改变用户归属或身份: %+v", key)
+		}
+	}
+	for _, query := range []struct {
+		sql, index string
+		args       []any
+	}{
+		{"SELECT id,source_db,puid,promote_id,idcard FROM tab_user WHERE source_db=? AND id IN ?", "idx_sw_user_report_identity", []any{"lx", ids}},
+		{"SELECT id,source_db,puid,promote_id,idcard FROM tab_user WHERE id IN ?", "idx_sw_user_report_identity", []any{ids}},
+		{"SELECT user_id,source_db,promote_id,game_id,server_id FROM tab_user_play_info WHERE server_id<>'' AND play_time BETWEEN ? AND ?", "idx_sw_play_report_active", []any{1, 501}},
+	} {
+		var plan []struct{ Key, Type, Extra string }
+		if err := db.Raw("EXPLAIN "+query.sql, query.args...).Scan(&plan).Error; err != nil {
+			t.Fatal(err)
+		}
+		if len(plan) != 1 || plan[0].Key != query.index || plan[0].Type != "range" || !strings.Contains(plan[0].Extra, "Using index") {
+			t.Fatalf("任务查询未使用覆盖范围索引: %+v", plan)
+		}
+	}
+	if err := db.Exec("ALTER TABLE tab_user DROP INDEX idx_sw_user_report_identity, ADD INDEX idx_sw_user_report_identity(id)").Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := migrateReportBuildQueryIndexes(db); err == nil || !strings.Contains(err.Error(), "idx_sw_user_report_identity") {
+		t.Fatalf("同名但列不符的索引必须拒绝: %v", err)
+	}
+}

internal/data/business/report_build.go

@@ -2,6 +2,7 @@ package business
 
 import (
 	"context"
+	"sort"
 	"strings"
 	"time"
 
@@ -111,12 +112,17 @@ func (r *reportBuildRepo) Rebuild(ctx context.Context, kind business.ReportBuild
 func loadReportInput(db *gorm.DB, kind business.ReportBuildKind, day time.Time) (*business.ReportInput, error) {
 	input := &business.ReportInput{Users: map[business.ReportUserKey]business.ReportUser{}, Games: map[business.AnalyticsKey]string{}}
 	begin, end := day.Unix(), day.AddDate(0, 0, 1).Unix()-1
-	userIDs, gameIDs := map[int64]bool{}, map[int64]bool{}
+	userKeys := map[business.ReportUserKey]bool{}
+	unknownUserIDs, gameIDs := map[int64]bool{}, map[int64]bool{}
 	collect := func(rows []reportEventPO, target *[]business.ReportEvent) {
 		for _, po := range rows {
 			*target = append(*target, toReportEvent(po))
 			if po.UserID != nil {
-				userIDs[*po.UserID] = true
+				if strings.TrimSpace(po.SourceDB) == "" {
+					unknownUserIDs[*po.UserID] = true
+				} else {
+					userKeys[business.ReportUserKey{SourceDB: po.SourceDB, ID: *po.UserID}] = true
+				}
 			}
 			if po.GameID != nil {
 				gameIDs[*po.GameID] = true
@@ -166,29 +172,16 @@ func loadReportInput(db *gorm.DB, kind business.ReportBuildKind, day time.Time)
 		return nil, business.ErrStorage
 	}
 	collect(pays, &input.Payments)
-	ids := make([]int64, 0, len(userIDs))
-	for id := range userIDs {
-		ids = append(ids, id)
+	var err error
+	input.Users, err = loadReportUsers(db, userKeys, unknownUserIDs)
+	if err != nil {
+		return nil, err
 	}
-	for start := 0; start < len(ids); start += 500 {
-		var users []struct {
-			ID              int64
-			SourceDB        string `gorm:"column:source_db"`
-			Puid, PromoteID *int64
-			IDCard          string `gorm:"column:idcard"`
-		}
-		if err := db.Table("tab_user").Select("id,source_db,puid,promote_id,idcard").Where("id IN ?", ids[start:min(start+500, len(ids))]).Scan(&users).Error; err != nil {
-			return nil, business.ErrStorage
-		}
-		for _, po := range users {
-			key := business.ReportUserKey{SourceDB: po.SourceDB, ID: po.ID}
-			input.Users[key] = business.ReportUser{ReportUserKey: key, ParentID: po.Puid, PromoteID: po.PromoteID, IDCard: po.IDCard}
-		}
-	}
-	ids = make([]int64, 0, len(gameIDs))
+	ids := make([]int64, 0, len(gameIDs))
 	for id := range gameIDs {
 		ids = append(ids, id)
 	}
+	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
 	for start := 0; start < len(ids); start += 500 {
 		var games []struct {
 			ID       int64
@@ -207,6 +200,61 @@ func loadReportInput(db *gorm.DB, kind business.ReportBuildKind, day time.Time)
 	return input, nil
 }
 
+// 已知平台的事件按完整主键读用户，避免将其他平台同号的大行一起回表。
+// 只有缺平台的事件读取全部同号用户，以保留旧逻辑的唯一主账号归属判断。
+func loadReportUsers(db *gorm.DB, known map[business.ReportUserKey]bool, unknown map[int64]bool) (map[business.ReportUserKey]business.ReportUser, error) {
+	result := map[business.ReportUserKey]business.ReportUser{}
+	read := func(query *gorm.DB) error {
+		var users []struct {
+			ID              int64
+			SourceDB        string `gorm:"column:source_db"`
+			Puid, PromoteID *int64
+			IDCard          string `gorm:"column:idcard"`
+		}
+		if err := query.Select("id,source_db,puid,promote_id,idcard").Scan(&users).Error; err != nil {
+			return business.ErrStorage
+		}
+		for _, po := range users {
+			key := business.ReportUserKey{SourceDB: po.SourceDB, ID: po.ID}
+			result[key] = business.ReportUser{ReportUserKey: key, ParentID: po.Puid, PromoteID: po.PromoteID, IDCard: po.IDCard}
+		}
+		return nil
+	}
+	bySource := make(map[string][]int64)
+	for key := range known {
+		if !unknown[key.ID] {
+			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)
+	// MySQL 5.7 的多列 IN 在部分批次退化为全表扫描；固定平台后
+	// 对 ID 做范围查询，仍完整匹配复合主键，并按 ID 顺序减少随机读页。
+	for _, source := range sources {
+		ids := bySource[source]
+		sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+		for start := 0; start < len(ids); start += 500 {
+			if err := read(db.Table("tab_user").Where("source_db=? AND id IN ?", source, ids[start:min(start+500, len(ids))])); err != nil {
+				return nil, err
+			}
+		}
+	}
+	ids := make([]int64, 0, len(unknown))
+	for id := range unknown {
+		ids = append(ids, id)
+	}
+	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+	for start := 0; start < len(ids); start += 500 {
+		if err := read(db.Table("tab_user").Where("id IN ?", ids[start:min(start+500, len(ids))])); err != nil {
+			return nil, err
+		}
+	}
+	return result, nil
+}
+
 func reportRegistrationQuery(db *gorm.DB) (*gorm.DB, error) {
 	q := db.Table("tab_user")
 	if db.Dialector.Name() != "mysql" || !db.Migrator().HasIndex("tab_user", "idx_sw_user_registration") {

internal/data/business/report_build_test.go

@@ -101,3 +101,73 @@ func TestReportBuildStorageReplacesBothTablesAtomically(t *testing.T) {
 		t.Fatal("失败后丢失原身份键或覆盖原汇总")
 	}
 }
+
+func TestReportUsersRestrictKnownPlatformsAndKeepAmbiguousMatches(t *testing.T) {
+	db := openTestDB(t)
+	if err := db.Exec("CREATE TABLE tab_user(id INTEGER,source_db TEXT,puid INTEGER,promote_id INTEGER,idcard TEXT,PRIMARY KEY(id,source_db))").Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Exec("INSERT INTO tab_user VALUES(1,'lx',0,10,'known'),(1,'hw',0,20,'unrelated'),(2,'lx',0,10,'ambiguous-a'),(2,'hw',0,20,'ambiguous-b'),(3,'yq',0,30,'unique')").Error; err != nil {
+		t.Fatal(err)
+	}
+	known := map[business.ReportUserKey]bool{{SourceDB: "lx", ID: 1}: true, {SourceDB: "lx", ID: 2}: true}
+	users, err := loadReportUsers(db, known, map[int64]bool{2: true, 3: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(users) != 4 {
+		t.Fatalf("wrong user scope: %+v", users)
+	}
+	if _, ok := users[business.ReportUserKey{SourceDB: "hw", ID: 1}]; ok {
+		t.Fatal("known source loaded another platform")
+	}
+	for _, key := range []business.ReportUserKey{{SourceDB: "lx", ID: 1}, {SourceDB: "lx", ID: 2}, {SourceDB: "hw", ID: 2}, {SourceDB: "yq", ID: 3}} {
+		if _, ok := users[key]; !ok {
+			t.Fatalf("lost identity needed for report/ambiguity: %+v", key)
+		}
+	}
+}
+
+func TestReportUsersAcrossBatches(t *testing.T) {
+	db := openTestDB(t)
+	if err := db.Exec("CREATE TABLE tab_user(id INTEGER,source_db TEXT,puid INTEGER,promote_id INTEGER,idcard TEXT,PRIMARY KEY(id,source_db))").Error; err != nil {
+		t.Fatal(err)
+	}
+	known := map[business.ReportUserKey]bool{}
+	unknown := map[int64]bool{}
+	rows := make([]map[string]any, 0, 2004)
+	for id := int64(1); id <= 1002; id++ {
+		for _, source := range []string{"lx", "hw"} {
+			rows = append(rows, map[string]any{"id": id, "source_db": source, "puid": 0, "promote_id": id, "idcard": source})
+		}
+		known[business.ReportUserKey{SourceDB: "lx", ID: id}] = true
+		if id > 501 {
+			unknown[id] = true
+		}
+	}
+	// 缺失用户不能被同号的其他平台用户替代。
+	known[business.ReportUserKey{SourceDB: "yq", ID: 1}] = true
+	if err := db.Table("tab_user").CreateInBatches(rows, 100).Error; err != nil {
+		t.Fatal(err)
+	}
+	users, err := loadReportUsers(db, known, unknown)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(users) != 1503 {
+		t.Fatalf("分批读取用户数量错误: %d", len(users))
+	}
+	for id := int64(1); id <= 1002; id++ {
+		user, ok := users[business.ReportUserKey{SourceDB: "lx", ID: id}]
+		if !ok || user.PromoteID == nil || *user.PromoteID != id || user.IDCard != "lx" {
+			t.Fatalf("跨批丢失或混淆用户资料: id=%d user=%+v", id, user)
+		}
+		_, other := users[business.ReportUserKey{SourceDB: "hw", ID: id}]
+		if other != unknown[id] {
+			t.Fatalf("跨批用户平台范围错误: id=%d", id)
+		}
+	}
+	if _, exists := users[business.ReportUserKey{SourceDB: "yq", ID: 1}]; exists {
+		t.Fatal("缺失平台用户不能回退到同号账号")
+	}
+}

internal/data/migrations_test.go

@@ -34,8 +34,8 @@ func TestMigrateAllRunsModuleSchemasWithoutBootstrapSeed(t *testing.T) {
 	if err = db.Table(migration.TableName).Count(&versions).Error; err != nil {
 		t.Fatal(err)
 	}
-	if versions != 27 {
-		t.Fatalf("migration versions = %d, want 27", versions)
+	if versions != 28 {
+		t.Fatalf("migration versions = %d, want 28", versions)
 	}
 	var communicationRows []dataintegration.ConfigPO
 	if err = db.Where("kind IN ?", []string{"mq", "websocket"}).Order("kind, provider").Find(&communicationRows).Error; err != nil {