package fixture import ( "context" "database/sql" "fmt" "sort" "strings" ) // 本文件是第二批的流程编排:Extract → Map → Load →(Verify)→(Idempotency)→ Clean。 // Batch2Report 是一次「搬 4 表」的结果。 type Batch2Report struct { Options Batch2Options LegacyCounts map[string]int LoadedCounts map[string]int RefUsers []int64 UsersWritten int MappingsWritten int Warnings []string DivideLogIDs []int64 BizNoSamples []BizNoSample } // BizNoSample 是 biz_no 口径的样本行(报告用,文本列已脱敏)。 type BizNoSample struct { Table string BizNo string Type int32 MoneyCents int64 CreateTime string RemarkMask string } // PrepareBatch2 把第二批 4 张表(+ 字典)搬进目标库。 // // 步骤: // 1. legacy 只读取出 4 张表的取样行 + 字典全表; // 2. 汇总这批数据里出现的**全部 legacy 用户引用**,一次性补 t_legacy_mapping + t_user; // 3. 逐行映射(batch2_mapping.go 纯函数)→ 幂等写入(batch2_load.go)。 func PrepareBatch2(ctx context.Context, cfg Config, opts Batch2Options) (*Batch2Report, 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 := &Batch2Report{Options: opts, LoadedCounts: map[string]int{}} // 落运行标记(同 batch1:create_time 的静默默认值只能靠断言抓) if _, err := WriteRunMarker(ctx, target, "batch2"); err != nil { return nil, err } divides, err := extractDivides(ctx, legacy, opts.DivideRootUserID) if err != nil { return nil, err } dlogs, err := extractDivideLogs(ctx, legacy, opts.DivideRootUserID) if err != nil { return nil, err } brs, err := extractBalanceRequests(ctx, legacy, opts.BalanceRequestUserID) if err != nil { return nil, err } elogs, err := extractExpenditureLogs(ctx, legacy, opts.ExpenditureLogUserID) if err != nil { return nil, err } if len(divides) == 0 || len(dlogs) == 0 || len(brs) == 0 || len(elogs) == 0 { return nil, fmt.Errorf( "取样不满足前置条件:divide_into=%d, divide_into_log=%d, balance_request=%d, expenditure_log=%d"+ "(4 张表都必须有行)", len(divides), len(dlogs), len(brs), len(elogs)) } var dicts []LegacyExpenditure if opts.WithExpenditureDict { dicts, err = extractExpenditures(ctx, legacy) if err != nil { return nil, err } if len(dicts) == 0 { return nil, fmt.Errorf("t_expenditure 字典为空,无法解析 expenditure_id") } } rep.LegacyCounts, err = countBatch2Legacy(ctx, legacy, opts) if err != nil { return nil, err } // ---- 1) 引用人映射行 ---- rep.RefUsers = collectUserRefs(divides, dlogs, brs, elogs) for _, uid := range rep.RefUsers { canonical := FixtureCanonicalUserID(uid) if err := upsertUser(ctx, target, canonical, "AC15 fixture b2(legacy="+itoa(uid)+")"); err != nil { return nil, err } if err := upsertMapping(ctx, target, MappingEntry{ CanonicalUserID: canonical, LegacyAcademicUserID: uid, InstitutionID: brs[0].SchoolID, CampusID: brs[0].CampusID, Role: "student", MatchSource: "seed", MatchConfidence: 100, }); err != nil { return nil, err } rep.UsersWritten++ rep.MappingsWritten++ } resolve := func(id int64) (string, bool) { if id <= 0 { return "", false } return FixtureCanonicalUserID(id), true } // ---- 2) 字典(D1,先写,供 expenditure_id 引用)---- for _, d := range dicts { td, w, err := MapExpenditure(d) if err != nil { return nil, err } rep.Warnings = append(rep.Warnings, w...) if err := upsertExpenditureDict(ctx, target, td); err != nil { return nil, err } } rep.LoadedCounts["t_expenditure"] = len(dicts) // ---- 3) 分账树 ---- for _, d := range divides { td, w, err := MapDivideInto(d, resolve) if err != nil { return nil, err } rep.Warnings = append(rep.Warnings, w...) if err := upsertDivideInto(ctx, target, td); err != nil { return nil, err } } rep.LoadedCounts["t_user_divide_into"] = len(divides) // ---- 4) 分账明细(D1)---- for _, l := range dlogs { tl, w, err := MapDivideIntoLog(l, resolve) if err != nil { return nil, err } rep.Warnings = append(rep.Warnings, w...) if err := upsertDivideIntoLog(ctx, target, tl); err != nil { return nil, err } rep.DivideLogIDs = append(rep.DivideLogIDs, tl.ID) } rep.LoadedCounts["t_user_divide_into_log"] = len(dlogs) // ---- 5) 充值工单(biz_no)---- for _, r := range brs { tr, w, err := MapBalanceRequest(r, resolve) if err != nil { return nil, err } rep.Warnings = append(rep.Warnings, w...) if err := upsertBalanceRequest(ctx, target, tr); err != nil { return nil, err } if len(rep.BizNoSamples) < 3 { rep.BizNoSamples = append(rep.BizNoSamples, BizNoSample{ Table: "t_user_balance_request", BizNo: tr.BizNo, Type: tr.Type, MoneyCents: tr.MoneyCents, CreateTime: tr.RequestTime.Format("2006-01-02 15:04:05"), RemarkMask: maskRemark(tr.Remark), }) } } rep.LoadedCounts["t_user_balance_request"] = len(brs) // ---- 6) 收支记录(biz_no)---- for _, e := range elogs { te, w, err := MapExpenditureLog(e, resolve) if err != nil { return nil, err } rep.Warnings = append(rep.Warnings, w...) if err := upsertExpenditureLog(ctx, target, te); err != nil { return nil, err } rep.BizNoSamples = append(rep.BizNoSamples, BizNoSample{ Table: "t_expenditure_log", BizNo: te.BizNo, Type: int32(te.ExpenditureID), MoneyCents: te.CostCents, CreateTime: te.CreateTime.Format("2006-01-02 15:04:05"), RemarkMask: maskRemark(te.Remark), }) } rep.LoadedCounts["t_expenditure_log"] = len(elogs) return rep, nil } // ---------- biz_no 快照与幂等比对 ---------- // BizNoSnapshot 记录 biz_no → target id 的映射(幂等比对的基线)。 type BizNoSnapshot map[string]map[string]int64 // table -> biz_no -> id // SnapshotBizNos 采集两张 biz_no 表当前的 (biz_no, id)。 func SnapshotBizNos(ctx context.Context, cfg Config) (BizNoSnapshot, error) { db, err := cfg.OpenTarget() if err != nil { return nil, err } defer db.Close() snap := BizNoSnapshot{} for _, t := range []string{"t_user_balance_request", "t_expenditure_log"} { rows, err := db.QueryContext(ctx, "SELECT biz_no, id FROM "+t) if err != nil { return nil, err } m := map[string]int64{} for rows.Next() { var b string var id int64 if err := rows.Scan(&b, &id); err != nil { rows.Close() return nil, err } m[b] = id } rows.Close() if err := rows.Err(); err != nil { return nil, err } snap[t] = m } return snap, nil } // IdempotencyResult 是「跑两次」的比对结论。 type IdempotencyResult struct { Before BizNoSnapshot After BizNoSnapshot SameRows map[string]bool // 行数不变 SameIDs map[string]bool // 同 biz_no 的 id 不变 NewBizNos map[string]int // 第二次跑新增了几条(必须为 0) IDListFirst map[string][]int64 IDListAgain map[string][]int64 } // AllOK 表示两张 biz_no 表都完全幂等。 func (r *IdempotencyResult) AllOK() bool { for _, t := range []string{"t_user_balance_request", "t_expenditure_log"} { if !r.SameRows[t] || !r.SameIDs[t] || r.NewBizNos[t] != 0 { return false } } return true } // IdempotencyCheck 先拍快照,再跑一次完全相同的映射,然后比对 —— // 这是 biz_no 口径「必须实测跑两次」的落地。 func IdempotencyCheck(ctx context.Context, cfg Config, opts Batch2Options) (*IdempotencyResult, error) { before, err := SnapshotBizNos(ctx, cfg) if err != nil { return nil, err } if _, err := PrepareBatch2(ctx, cfg, opts); err != nil { return nil, fmt.Errorf("第二次映射失败(幂等被破坏):%w", err) } after, err := SnapshotBizNos(ctx, cfg) if err != nil { return nil, err } res := &IdempotencyResult{ Before: before, After: after, SameRows: map[string]bool{}, SameIDs: map[string]bool{}, NewBizNos: map[string]int{}, IDListFirst: map[string][]int64{}, IDListAgain: map[string][]int64{}, } for _, t := range []string{"t_user_balance_request", "t_expenditure_log"} { b, a := before[t], after[t] res.SameRows[t] = len(b) == len(a) same := true for bizNo, id := range b { if a[bizNo] != id { same = false } } res.SameIDs[t] = same newCount := 0 for bizNo := range a { if _, ok := b[bizNo]; !ok { newCount++ } } res.NewBizNos[t] = newCount for _, id := range b { res.IDListFirst[t] = append(res.IDListFirst[t], id) } for _, id := range a { res.IDListAgain[t] = append(res.IDListAgain[t], id) } sort.Slice(res.IDListFirst[t], func(i, j int) bool { return res.IDListFirst[t][i] < res.IDListFirst[t][j] }) sort.Slice(res.IDListAgain[t], func(i, j int) bool { return res.IDListAgain[t][i] < res.IDListAgain[t][j] }) } return res, nil } // ---------- 清理 ---------- // Batch2CleanReport 是清理结果。 type Batch2CleanReport struct { Deleted map[string]int64 Remaining map[string]int64 BizNos int } // CleanupBatch2 删除第二批 fixture 造的全部数据。 // // 删除范围严格按「这批取样出来的 biz_no / legacy id / 引用人 ULID」限定,不误删。 func CleanupBatch2(ctx context.Context, cfg Config, opts Batch2Options) (*Batch2CleanReport, 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() divides, err := extractDivides(ctx, legacy, opts.DivideRootUserID) if err != nil { return nil, err } dlogs, err := extractDivideLogs(ctx, legacy, opts.DivideRootUserID) if err != nil { return nil, err } brs, err := extractBalanceRequests(ctx, legacy, opts.BalanceRequestUserID) if err != nil { return nil, err } elogs, err := extractExpenditureLogs(ctx, legacy, opts.ExpenditureLogUserID) if err != nil { return nil, err } rep := &Batch2CleanReport{Deleted: map[string]int64{}, Remaining: map[string]int64{}} // 1) biz_no 表:按 biz_no 精确删除 del := func(table string, cond string, args []any) error { res, err := target.ExecContext(ctx, fmt.Sprintf("DELETE FROM %s WHERE %s", table, cond), args...) if err != nil { return fmt.Errorf("清理 %s 失败:%w", table, err) } n, _ := res.RowsAffected() rep.Deleted[table] += n return nil } for _, r := range brs { if err := del("t_user_balance_request", "biz_no = ?", []any{r.BizNo}); err != nil { return nil, err } } for _, e := range elogs { if err := del("t_expenditure_log", "biz_no = ?", []any{e.BizNo}); err != nil { return nil, err } } rep.BizNos = len(brs) + len(elogs) // 变异实验残留清理:只删「属于本批 fixture 用户、但 biz_no 不是 legacy 原值」的行。 // 存在的理由:把 biz_no 改成 UUID 的变异(M1)会插进一批 biz_no 错误的新行, // 按正确 biz_no 删不掉;这里用「fixture 用户 + 非期望 biz_no」这个双重条件兜住, // 既不会漏,也不会误删任何非 fixture 行。 refsForResidue := collectUserRefs(divides, dlogs, brs, elogs) canonicals := make([]any, 0, len(refsForResidue)) for _, uid := range refsForResidue { canonicals = append(canonicals, FixtureCanonicalUserID(uid)) } if len(canonicals) > 0 { brNos := make([]any, 0, len(brs)) for _, r := range brs { brNos = append(brNos, r.BizNo) } elNos := make([]any, 0, len(elogs)) for _, e := range elogs { elNos = append(elNos, e.BizNo) } inCanon := placeholders(len(canonicals)) if err := del("t_user_balance_request", fmt.Sprintf("user_id IN (%s) AND biz_no NOT IN (%s)", inCanon, placeholders(len(brNos))), append(append([]any{}, canonicals...), brNos...)); err != nil { return nil, err } if err := del("t_expenditure_log", fmt.Sprintf("create_user_id IN (%s) AND biz_no NOT IN (%s)", inCanon, placeholders(len(elNos))), append(append([]any{}, canonicals...), elNos...)); err != nil { return nil, err } } // 2) D1 表:按 legacy id 删除 for _, l := range dlogs { if err := del("t_user_divide_into_log", "id = ?", []any{l.ID}); err != nil { return nil, err } } // 3) 分账树:按 ULID 删除 for _, d := range divides { if err := del("t_user_divide_into", "user_id = ?", []any{FixtureCanonicalUserID(d.UserID)}); err != nil { return nil, err } } // 4) 字典:只删这批 D1 进去的 id(其余行不动) if opts.WithExpenditureDict { dicts, err := extractExpenditures(ctx, legacy) if err != nil { return nil, err } for _, d := range dicts { if err := del("t_expenditure", "id = ?", []any{d.ID}); err != nil { return nil, err } } } // 5) 引用人映射 + 用户行 refs := collectUserRefs(divides, dlogs, brs, elogs) for _, uid := range refs { canonical := FixtureCanonicalUserID(uid) if err := del("t_legacy_mapping", "canonical_user_id = ?", []any{canonical}); err != nil { return nil, err } if err := del("t_user", "canonical_user_id = ?", []any{canonical}); err != nil { return nil, err } } // 5.5) 运行标记也清掉 if _, err := ClearRunMarkers(ctx, target); err != nil { return nil, err } // 6) 复核 5 张表**全表**行数 for _, t := range []string{"t_expenditure", "t_user_divide_into", "t_user_divide_into_log", "t_user_balance_request", "t_expenditure_log", "t_legacy_mapping", "t_user"} { var n int64 if err := target.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+t).Scan(&n); err != nil { return nil, err } rep.Remaining[t] = n } return rep, nil } // placeholders 生成 n 个 "?" 的占位符串。 func placeholders(n int) string { if n <= 0 { return "NULL" } out := make([]string, n) for i := range out { out[i] = "?" } return strings.Join(out, ",") } // CountBatch2Tables 统计这批涉及的表的当前行数。 func CountBatch2Tables(ctx context.Context, cfg Config) (map[string]int64, error) { db, err := cfg.OpenTarget() if err != nil { return nil, err } defer db.Close() out := map[string]int64{} for _, t := range []string{"t_expenditure", "t_user_divide_into", "t_user_divide_into_log", "t_user_balance_request", "t_expenditure_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 } // fetchDivideIntoIDs 取分账明细的 target id 集合(D1 一致性对比用)。 func fetchDivideIntoIDs(ctx context.Context, db *sql.DB) ([]int64, error) { rows, err := db.QueryContext(ctx, "SELECT id FROM t_user_divide_into_log ORDER BY id") if err != nil { return nil, err } defer rows.Close() var out []int64 for rows.Next() { var id int64 if err := rows.Scan(&id); err != nil { return nil, err } out = append(out, id) } return out, rows.Err() }