package fixture import ( "context" "database/sql" "fmt" "sort" "strings" ) // 本文件是流程编排:Pick → Extract → Map → Load →(Verify)→ Clean。 // // 每一步都可单独从 CLI 调用,因此**没有任何跨进程的内存状态**: // canonical_user_id 由 FixtureCanonicalUserID 确定性派生,重复执行天然幂等。 // MapReport 是一次「搬数据」的结果。 type MapReport struct { LegacyUserID int64 CanonicalUserID string LegacyCounts map[string]int LegacyRows int Warnings []string CreatorIDs []int64 UsersWritten int MappingsWritten int LogIDs []int64 // 样本:3 条流水(id/type/money/时间),供报告引用(remark 已脱敏)。 Samples []LogSample } // LogSample 是报告里引用的流水样本。 type LogSample struct { ID int64 Type int32 Money string CreateTime string RemarkMask string } // PrepareAndLoad 把选中用户的一条链路搬到目标库。 // // 步骤: // 1. legacy 侧只读取出 1 账户 + N 校区账户 + M 条流水; // 2. 为「本人 + 流水里出现的创建人」各插 1 行 t_legacy_mapping 与 t_user; // —— 创建人是 create_user_id 的引用对象,不插就会退化成 NULL,核对就看不到「解析成功」这一半; // 3. 逐行映射(mapping.go 纯函数)→ 幂等写入(load.go)。 func PrepareAndLoad(ctx context.Context, cfg Config, legacyUserID int64) (*MapReport, error) { legacy, err := cfg.OpenLegacy() if err != nil { return nil, err } defer legacy.Close() target, err := cfg.OpenTarget() if err != nil { return nil, err } defer target.Close() rep := &MapReport{LegacyUserID: legacyUserID} expense, err := extractExpense(ctx, legacy, legacyUserID) if err != nil { return nil, err } schools, err := extractExpenseSchools(ctx, legacy, legacyUserID) if err != nil { return nil, err } logs, err := extractExpenseLogs(ctx, legacy, legacyUserID) if err != nil { return nil, err } if len(schools) == 0 { return nil, fmt.Errorf("legacy 用户 %d 无 t_user_expense_school 行,不满足 fixture 前置条件", legacyUserID) } if len(logs) < 3 { return nil, fmt.Errorf("legacy 用户 %d 只有 %d 条流水(要求 >=3)", legacyUserID, len(logs)) } rep.LegacyCounts, err = countLegacyRows(ctx, legacy, legacyUserID) if err != nil { return nil, err } rep.LegacyRows = rep.LegacyCounts["t_user_expense"] + rep.LegacyCounts["t_user_expense_school"] + rep.LegacyCounts["t_user_expense_log"] canonical := FixtureCanonicalUserID(legacyUserID) rep.CanonicalUserID = canonical // ---- 创建人集合(去重、升序)---- creatorSet := map[int64]bool{} for _, l := range logs { if l.CreateUserID.Valid && l.CreateUserID.Int64 > 0 { creatorSet[l.CreateUserID.Int64] = true } } for id := range creatorSet { rep.CreatorIDs = append(rep.CreatorIDs, id) } sort.Slice(rep.CreatorIDs, func(i, j int) bool { return rep.CreatorIDs[i] < rep.CreatorIDs[j] }) // ---- 1) t_legacy_mapping + t_user 行 ---- if err := upsertMapping(ctx, target, MappingEntry{ CanonicalUserID: canonical, LegacyAcademicUserID: legacyUserID, InstitutionID: schools[0].SchoolID, CampusID: schools[0].CampusID, Role: "student", MatchSource: "seed", MatchConfidence: 100, }); err != nil { return nil, err } rep.MappingsWritten++ if err := upsertUser(ctx, target, canonical, "AC15 fixture 账户(legacy="+itoa(legacyUserID)+")"); err != nil { return nil, err } rep.UsersWritten++ for _, cid := range rep.CreatorIDs { uid := FixtureCanonicalUserID(cid) if err := upsertMapping(ctx, target, MappingEntry{ CanonicalUserID: uid, LegacyAcademicUserID: cid, InstitutionID: schools[0].SchoolID, CampusID: schools[0].CampusID, Role: "staff", MatchSource: "seed", MatchConfidence: 100, }); err != nil { return nil, err } rep.MappingsWritten++ if err := upsertUser(ctx, target, uid, "AC15 fixture 创建人(legacy="+itoa(cid)+")"); err != nil { return nil, err } rep.UsersWritten++ } resolver := func(id int64) (string, bool) { if id <= 0 { return "", false } return FixtureCanonicalUserID(id), true } // ---- 2) 全局账户 ---- te, warns, err := MapExpense(*expense, canonical) if err != nil { return nil, err } rep.Warnings = append(rep.Warnings, warns...) if err := upsertExpense(ctx, target, te); err != nil { return nil, err } // ---- 3) 校区账户 ---- for _, s := range schools { ts, w, err := MapExpenseSchool(s, canonical) if err != nil { return nil, err } rep.Warnings = append(rep.Warnings, w...) if err := upsertExpenseSchool(ctx, target, ts); err != nil { return nil, err } } // ---- 4) 流水(D1 显式 id)---- for _, l := range logs { tl, w, err := MapExpenseLog(l, canonical, resolver) if err != nil { return nil, err } rep.Warnings = append(rep.Warnings, w...) if err := insertExpenseLog(ctx, target, tl); err != nil { return nil, err } rep.LogIDs = append(rep.LogIDs, tl.ID) } // ---- 5) 样本(前 3 条,remark 脱敏)---- for i, l := range logs { if i >= 3 { break } rep.Samples = append(rep.Samples, LogSample{ ID: l.ID, Type: l.Type, Money: l.Money.String, CreateTime: l.CreateTime.Time.Format("2006-01-02 15:04:05"), RemarkMask: maskRemark(l.Remark), }) } return rep, nil } // CleanReport 是清理结果。 type CleanReport struct { Deleted map[string]int64 Remaining map[string]int64 CanonicalIDs []string } // Cleanup 删除本 fixture 造的全部数据,并回报 4 张表的剩余行数。 // // 删除范围严格限定在「由 FixtureCanonicalUserID 派生出来的 ULID」上, // 不会误删任何非 fixture 行。 func Cleanup(ctx context.Context, cfg Config, legacyUserID int64) (*CleanReport, error) { target, err := cfg.OpenTarget() if err != nil { return nil, err } defer target.Close() legacy, err := cfg.OpenLegacy() if err != nil { return nil, err } defer legacy.Close() logs, err := extractExpenseLogs(ctx, legacy, legacyUserID) if err != nil { return nil, err } creatorSet := map[int64]bool{legacyUserID: true} for _, l := range logs { if l.CreateUserID.Valid && l.CreateUserID.Int64 > 0 { creatorSet[l.CreateUserID.Int64] = true } } var ids []int64 for id := range creatorSet { ids = append(ids, id) } sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) var canonicals []string for _, id := range ids { canonicals = append(canonicals, FixtureCanonicalUserID(id)) } rep := &CleanReport{Deleted: map[string]int64{}, Remaining: map[string]int64{}, CanonicalIDs: canonicals} inList := strings.TrimSuffix(strings.Repeat("?,", len(canonicals)), ",") args := make([]any, len(canonicals)) for i, c := range canonicals { args[i] = c } // 1) 探针行(变异/异常中断时可能残留) if _, err := target.ExecContext(ctx, `DELETE FROM t_user_expense_log WHERE remark = 'ac15 fixture auto-inc probe'`); err != nil { return nil, err } for _, t := range []string{"t_user_expense_log", "t_user_expense_school", "t_user_expense"} { res, err := target.ExecContext(ctx, fmt.Sprintf("DELETE FROM %s WHERE user_id IN (%s)", t, inList), args...) if err != nil { return nil, fmt.Errorf("清理 %s 失败:%w", t, err) } n, _ := res.RowsAffected() rep.Deleted[t] = n } // 映射与用户行 res, err := target.ExecContext(ctx, fmt.Sprintf("DELETE FROM t_legacy_mapping WHERE canonical_user_id IN (%s)", inList), args...) if err != nil { return nil, err } n, _ := res.RowsAffected() rep.Deleted["t_legacy_mapping"] = n res, err = target.ExecContext(ctx, fmt.Sprintf("DELETE FROM t_user WHERE canonical_user_id IN (%s)", inList), args...) if err != nil { return nil, err } n, _ = res.RowsAffected() rep.Deleted["t_user"] = n // 复核 4 张表的**全表**行数(不按 user 过滤:要证明回到 0 行) for _, t := range []string{"t_user_expense", "t_user_expense_school", "t_user_expense_log"} { var cnt int64 if err := target.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+t).Scan(&cnt); err != nil { return nil, err } rep.Remaining[t] = cnt } for _, t := range []string{"t_legacy_mapping", "t_user"} { var cnt int64 if err := target.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+t).Scan(&cnt); err != nil { return nil, err } rep.Remaining[t] = cnt } return rep, nil } // CountTargetTables 统计目标 4 张表当前行数(清理前后对照用)。 func CountTargetTables(ctx context.Context, cfg Config) (map[string]int64, error) { db, err := cfg.OpenTarget() if err != nil { return nil, err } defer db.Close() return countTargetTablesWith(ctx, db) } func countTargetTablesWith(ctx context.Context, db *sql.DB) (map[string]int64, error) { out := map[string]int64{} for _, t := range []string{"t_user_expense", "t_user_expense_school", "t_user_expense_log", "t_legacy_mapping", "t_user"} { var n int64 if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+t).Scan(&n); err != nil { return nil, err } out[t] = n } return out, nil } func itoa(v int64) string { return fmt.Sprintf("%d", v) }