DEV-MERGE-LEGACY-BACKEND-2026/internal/academic/dualrun/fixture/batch2_run.go

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()
}