From 34535c90c3b4d7dc09ac4441dbdcb85b2cada567 Mon Sep 17 00:00:00 2001 From: Yvan <8574526@qq,com> Date: Wed, 4 Nov 2026 15:46:46 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E5=95=86=E5=8A=A1?= =?UTF-8?q?=E6=B1=87=E6=80=BB=E4=BB=BB=E5=8A=A1=E8=B6=85=E6=97=B6=E5=B9=B6?= =?UTF-8?q?=E9=AA=8C=E8=AF=81=E8=AE=A1=E5=88=92=E5=90=AF=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/sw6-to-kra/report-build-source-notes.md | 8 +- docs/sw6-to-kra/validation.md | 35 ++++++-- internal/data/business/migrations.go | 2 +- internal/data/business/query_indexes.go | 15 +++- internal/data/business/query_indexes_test.go | 68 +++++++++++++++ internal/data/business/report_build.go | 90 +++++++++++++++----- internal/data/business/report_build_test.go | 70 +++++++++++++++ internal/data/migrations_test.go | 4 +- internal/modules/catalog_test.go | 4 +- 9 files changed, 260 insertions(+), 36 deletions(-) diff --git a/docs/sw6-to-kra/report-build-source-notes.md b/docs/sw6-to-kra/report-build-source-notes.md index a9cee09..e57dfe0 100644 --- a/docs/sw6-to-kra/report-build-source-notes.md +++ b/docs/sw6-to-kra/report-build-source-notes.md @@ -10,7 +10,7 @@ - 新增付费先取同一分组内“当日注册账号 ∩ 当日付费账号”，再转身份证；不能直接用身份证交集，否则会把同证件的其他账号误认成新增付费。 - 区服注册取 `tab_user_play_info.create_time` 的创角，活跃取 `play_time`，不套用日注册/登录的主账号限制。区服编号去首尾空白，保留前导零；创角/活跃缺游戏 ID 时忽略，充值缺游戏 ID 沿用 0。区服汇总没有新增付费指标。 - 两类报表均按日、平台、渠道、游戏（区服额外编号）生成模式 0/1 两行及对应身份键。模式 1 去首尾空格、仅将末尾 X 转 x，空身份证丢弃。金额两种模式相同，查询层仍固定读取模式 0。 -- 游戏名作为当前游戏资料的快照写入。游戏/用户按关联 ID 批量加载，平台作为字典键的一部分，不混淆跨平台同号。 +- 游戏名作为当前游戏资料的快照写入。已知平台的用户按平台分组、ID 排序后分批加载；缺平台的事件仍读取全部同号用户以判断唯一主账号。平台作为字典键的一部分，不混淆跨平台同号。 ## 原子替换与任务 @@ -18,7 +18,7 @@ - 汇总、身份键按批写入，失败或取消回滚原有分区；其他日期不变。输出日期固定 `yyyy-MM-dd`，不采用驱动返回的时间格式作为写入日期。 - 任务方法保留 `swReportDayTodayTask` / `YesterdayTask` / `BackfillTask` 和对应 `swReportServerDay*` 六个名字。已有迁移 SQL 保留旧执行器名和 `_execution` 参数，可直接绑定 Kra 方法执行器。 - 回跑参数为 startDate/endDate；每个缺省日期独立取昨日，闭区间逐日重建、日间暂停一秒，最后一天不暂停。支持上下文取消，非法/倒置日期报错。 -- 不新建默认商务计划、不启用旧导入计划。原始同步表和四张业务报表表沿用旧库结构；本批只新建日期互斥锁表，不在系统库中伪造完整业务数据源。 +- 迁移本身不新建默认商务计划、不启用旧导入计划；实际运行状态由管理员单独配置。原始同步表和四张业务报表表沿用旧库结构，不在系统库中伪造完整业务数据源。 ## 验证边界 @@ -59,4 +59,6 @@ 普通渠道列表、注册明细、充值明细、渠道激活、工作台未登录榜，以及游戏/区服下拉选项不直接使用这四张汇总表。预警检测 `alertTriggerTask` 另行读取原始注册/充值数据，不由这些汇总任务生成预警。 -今天任务用于当天持续刷新，昨天任务用于补算昨日完整数据，backfill 用于补历史或纠正迟到数据。2026-11-04 核实时六条计划全部停用：不自动刷新汇总，但已有历史结果仍可查询；若原始数据继续同步，依赖汇总的指标会与明细逐渐产生时间差。 +今天任务用于当天持续刷新，昨天任务用于补算昨日完整数据，backfill 用于补历史或纠正迟到数据。2026-11-04 按用户要求与旧计划同步：四条今天/昨天汇总计划和预警计划启用；两个 backfill 保持停用，需要时手动运行。周期、时区和超时/重试参数保持原配置，详见 [验收记录](validation.md)。 + +报表输入索引由版本 `202611040014_business_report_build_indexes` 增量添加：`tab_user.idx_sw_user_report_identity` 覆盖用户身份及渠道字段，`tab_user_play_info.idx_sw_play_report_active` 覆盖活跃时间查询的全部读取字段。迁移跳过未接入的业务表，重复执行不重建已有同列索引，同名异列索引报错。MySQL 使用 `ALGORITHM=INPLACE, LOCK=NONE`，不改写原始业务字段。 diff --git a/docs/sw6-to-kra/validation.md b/docs/sw6-to-kra/validation.md index 59ba8c9..dc06175 100644 --- a/docs/sw6-to-kra/validation.md +++ b/docs/sw6-to-kra/validation.md @@ -28,15 +28,38 @@ | 方法 | 周期 | 测试库状态 | |---|---|---| -| swReportDayYesterdayTask | 每日 01:00 | 停用 | -| swReportServerDayYesterdayTask | 每日 01:05 | 停用 | -| swReportDayTodayTask | 每 5 分钟，第 40 秒 | 停用 | -| swReportServerDayTodayTask | 每 5 分钟 | 停用 | +| swReportDayYesterdayTask | 每日 01:00 | 启用 | +| swReportServerDayYesterdayTask | 每日 01:05 | 启用 | +| swReportDayTodayTask | 每 5 分钟，第 40 秒 | 启用 | +| swReportServerDayTodayTask | 每 5 分钟 | 启用 | | swReportDayBackfillTask | 每 60 秒，原配置停用 | 停用 | | swReportServerDayBackfillTask | 每 60 秒，原配置停用 | 停用 | -| alertTriggerTask | 每 30 分钟 | 停用 | +| alertTriggerTask | 每 30 分钟 | 启用 | -新库未开启自动执行；框架自带 ClearDB、CleanStaleUploads 另行保留。旧工程的演示、分片、MapReduce 示例没有实际商务计划，不属于漏迁。方法超时后仍未退出时禁止重复执行，同分区报表和预警写入有事务锁及去重测试。 +2026-11-04 按用户确认与旧库启停状态同步，五条常规任务开启自动执行，两个历史回填保持停用；框架自带 ClearDB、CleanStaleUploads 继续停用。七条计划的周期、启停、上海时区、下次执行时间及超时/重试参数逐条核对通过，常规任务仍为 60 秒超时。旧工程的演示、分片、MapReduce 示例没有实际商务计划，不属于漏迁。方法超时后仍未退出时禁止重复执行，同分区报表和预警写入有事务锁及去重测试。 + +### 启用后的实际复测 + +中断续接时，昨日两类汇总均有 60 秒超时记录。用户查询改为按平台分组、ID 排序分批，避开 MySQL 5.7 多列 IN 部分批次的全表扫描；新增用户身份和区服活跃覆盖索引，减少大行回表。原始业务字段、统计规则和任务执行参数不变。 + +版本 `202611040014_business_report_build_indexes` 已在 sw9 应用，迁移版本总数为 28。两条索引分别为 `tab_user.idx_sw_user_report_identity(id,source_db,puid,promote_id,idcard)`、`tab_user_play_info.idx_sw_play_report_active(play_time,user_id,source_db,promote_id,game_id,server_id)`。 + +| 复测项 | 结果 | +|---|---| +| 昨日商务汇总，统计日 2026-11-03 | 最终版本实际执行成功，21.032 秒 | +| 昨日区服汇总，统计日 2026-11-03 | 最终版本实际执行成功，20.243 秒 | +| 今日两个汇总 | 已观察到多轮自动成功执行 | +| 预警任务 | 15:30 自动成功，28.864 秒；两库 480 条业务记录相等，用户 ID 按迁移表映射，排除生成 ID 和创建/更新时间 | +| 区服用户资料读取 | 同批 18,177 用户，71.619 秒降至 0.443 秒，38 批全部使用覆盖索引 | +| 区服活跃输入读取 | 同批 20,564 行，28.545 秒降至 0.074 秒，使用覆盖范围索引 | + +以上为该测试库的单轮实测，包含当时缓存和数据库负载影响，不是固定延迟承诺。Go 全量测试、构建及独立 MySQL 5.7 的六组回归通过；新增回归验证覆盖索引、批量身份读取、索引幂等和同名异列冲突拒绝，既有事务回滚测试通过。 + +四张表的 2026-11-03 重建结果分别为 3,440、15,064、19,682、30,671 行，合计 **68,857 行**，逐行及 SHA-256 均与此前保存的旧程序同日重建结果一致。该比较明确复用旧重算基线，本轮没有重跑或覆盖 sw8 的昨日分区。 + +直接比较两库当前昨日分区仍有差异：sw8 保留此前恢复的旧汇总，日报 8 行指标变化、日报身份键少 10 行，区服汇总 15 行旧结果对应 29 行重建结果、区服身份键少 31 行。不能把重算基线一致表述为两库当前分区一致。2026-11-04 两库四表均为空，sw9 当日注册、登录、成功充值、创角及活跃原始记录均为 0；今日验证覆盖自动触发和空日处理，非空数据核对使用昨日。 + +原始对比、任务日志和查询计划位于仓库外 `D:/Users/Yvan/backups/sw9/20261104_rehearsal/tasks-enabled-20261104_145308/`，最终证据为 `resumed-verification.json`、`final-manual-task-logs.json`、`task-index-installation.json`。辅助脚本同样留在仓库外，未恢复项目内已删除的 scripts 目录。最终版本运行于日常 8000 服务，保持五条自动计划启用。 ## 已完成的性能优化 diff --git a/internal/data/business/migrations.go b/internal/data/business/migrations.go index 005ae18..fe123cb 100644 --- a/internal/data/business/migrations.go +++ b/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{}) diff --git a/internal/data/business/query_indexes.go b/internal/data/business/query_indexes.go index 0fbd334..36d22fa 100644 --- a/internal/data/business/query_indexes.go +++ b/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) } diff --git a/internal/data/business/query_indexes_test.go b/internal/data/business/query_indexes_test.go index a6c9e9a..2c28ed6 100644 --- a/internal/data/business/query_indexes_test.go +++ b/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) + } +} diff --git a/internal/data/business/report_build.go b/internal/data/business/report_build.go index 85cbdce..a0f69bb 100644 --- a/internal/data/business/report_build.go +++ b/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") { diff --git a/internal/data/business/report_build_test.go b/internal/data/business/report_build_test.go index c052801..ce885c6 100644 --- a/internal/data/business/report_build_test.go +++ b/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("缺失平台用户不能回退到同号账号") + } +} diff --git a/internal/data/migrations_test.go b/internal/data/migrations_test.go index 2048b40..3c2aff2 100644 --- a/internal/data/migrations_test.go +++ b/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 { diff --git a/internal/modules/catalog_test.go b/internal/modules/catalog_test.go index b319bde..6518730 100644 --- a/internal/modules/catalog_test.go +++ b/internal/modules/catalog_test.go @@ -25,8 +25,8 @@ func TestCatalogContainsBuiltInModulesInDependencyOrder(t *testing.T) { t.Fatalf("definition[%d] = %q, want %q", index, got, name) } } - if got := catalog.MigrationSteps(); len(got) != 27 { - t.Fatalf("module migrations = %d, want 27", len(got)) + if got := catalog.MigrationSteps(); len(got) != 28 { + t.Fatalf("module migrations = %d, want 28", len(got)) } var menus, apis int for _, definition := range catalog.Definitions {