package fixture

import (
	"context"
	"database/sql"
	"fmt"
	"sort"
	"strconv"
	"strings"
)

// 本文件是第二批的**逐列核对**。两条与上一批相同的铁律：
//  1. 期望值**不由本包映射代码产生**：金额/时间一律由 MySQL 服务端
//     `CAST(... AS DECIMAL(12,2))` / `DATE_FORMAT(...)` 算出；
//  2. 用户引用（ULID）的期望值来自**目标库 t_legacy_mapping 的实际数据**
//     （反向 JOIN 查出来），而不是调用 FixtureCanonicalUserID —— 否则映射写错时两边一起错。
//
// 另外本批专门验证 biz_no 四项（原样保留 / 幂等 / id 解耦 / 反查可用）。

// ColumnAgg 是「一列 × N 行」的聚合核对结果（避免 600+ 行逐格输出）。
type ColumnAgg struct {
	Table   string
	Column  string
	Total   int
	Equal   int
	Samples []string // 最多 3 条不一致样例
	Note    string
}

// OK 表示该列所有行都相等。
func (c ColumnAgg) OK() bool { return c.Total > 0 && c.Equal == c.Total }

// colAcc 是聚合累加器。
type colAcc struct {
	table, column, note string
	total, equal        int
	samples             []string
}

func newCol(table, column, note string) *colAcc {
	return &colAcc{table: table, column: column, note: note}
}

// cmp 比一行；不一致时留样例（最多 3 条）。
func (a *colAcc) cmp(rowKey, legacy, target string, ok bool) {
	a.total++
	if ok {
		a.equal++
		return
	}
	if len(a.samples) < 3 {
		a.samples = append(a.samples, fmt.Sprintf("%s: legacy=%q target=%q", rowKey, truncate(legacy, 28), truncate(target, 28)))
	}
}

func (a *colAcc) done() ColumnAgg {
	return ColumnAgg{Table: a.table, Column: a.column, Total: a.total, Equal: a.equal,
		Samples: a.samples, Note: a.note}
}

// BizNoVerify 是 biz_no 四项验证的结果。
type BizNoVerify struct {
	Table            string
	Rows             int
	CharEqual        int  // 逐字符相等的行数
	Len32            int  // 长度为 32 的行数
	HexOK            int  // 32 位小写 hex 的行数
	UniqueBizNo      int  // DISTINCT biz_no 数
	IDEqualsPrefix   bool // id 是否等于 biz_no 派生的数字（**必须为 false**，证明未做反算）
	ReverseJoin      int  // 用 biz_no JOIN 回 legacy 的行数
	LegacyRows       int
	MirrorOK         bool     // 反序插入镜像：id 跟插入顺序、不跟 biz_no
	MirrorSample     []string // 镜像里 (id, biz_no) 前 5 行
	AutoIncrementVal int64
	MaxID            int64
}

// D1Verify 是 D1 口径的 id 一致性结果。
type D1Verify struct {
	Table     string
	LegacyIDs []int64
	TargetIDs []int64
	Pass      bool
}

// Batch2VerifyReport 是整轮核对结果。
type Batch2VerifyReport struct {
	Options    Batch2Options
	Cols       []ColumnAgg
	Assertions []Assertion
	BizNo      []BizNoVerify
	D1         []D1Verify
	TargetRows map[string]int64
	Warnings   []string
}

// FailCount 统计不一致列数 + 失败断言数。
func (r *Batch2VerifyReport) FailCount() int {
	n := 0
	for _, c := range r.Cols {
		if !c.OK() {
			n++
		}
	}
	for _, a := range r.Assertions {
		if !a.Pass {
			n++
		}
	}
	return n
}

// OK 表示整轮通过。
func (r *Batch2VerifyReport) OK() bool { return r.FailCount() == 0 }

// ulidIndex 从目标库 t_legacy_mapping 反查 legacy int → ULID（数据驱动的期望值来源）。
func ulidIndex(ctx context.Context, db *sql.DB) (map[int64]string, error) {
	rows, err := db.QueryContext(ctx,
		`SELECT legacy_academic_user_id, canonical_user_id FROM t_legacy_mapping
		  WHERE legacy_academic_user_id IS NOT NULL`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := map[int64]string{}
	for rows.Next() {
		var id int64
		var uid string
		if err := rows.Scan(&id, &uid); err != nil {
			return nil, err
		}
		out[id] = uid
	}
	return out, rows.Err()
}

// expectedULID 用映射表数据解析期望 ULID；0/NULL/无映射 → "NULL"。
func expectedULID(idx map[int64]string, v sql.NullInt64) string {
	if !v.Valid || v.Int64 <= 0 {
		return "NULL"
	}
	if uid, ok := idx[v.Int64]; ok {
		return uid
	}
	return "UNMAPPED(" + strconv.FormatInt(v.Int64, 10) + ")"
}

func nullStrOrNull(v sql.NullString) string {
	if !v.Valid {
		return "NULL"
	}
	return v.String
}

// VerifyBatch2 执行第二批的完整核对。
func VerifyBatch2(ctx context.Context, cfg Config, opts Batch2Options) (*Batch2VerifyReport, 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()

	idx, err := ulidIndex(ctx, target)
	if err != nil {
		return nil, err
	}

	rep := &Batch2VerifyReport{Options: opts, TargetRows: 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"} {
		var n int64
		if err := target.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+t).Scan(&n); err != nil {
			return nil, err
		}
		rep.TargetRows[t] = n
	}

	if err := verifyDivideInto(ctx, legacy, target, opts, idx, rep); err != nil {
		return nil, err
	}
	if err := verifyDivideIntoLog(ctx, legacy, target, opts, idx, rep); err != nil {
		return nil, err
	}
	if err := verifyBalanceRequest(ctx, legacy, target, opts, idx, rep); err != nil {
		return nil, err
	}
	if err := verifyExpenditureLog(ctx, legacy, target, opts, idx, rep); err != nil {
		return nil, err
	}
	if opts.WithExpenditureDict {
		if err := verifyExpenditureDict(ctx, legacy, target, rep); err != nil {
			return nil, err
		}
	}
	if err := verifyBizNoFour(ctx, legacy, target, opts, rep); err != nil {
		return nil, err
	}
	return rep, nil
}

// ---------- t_user_divide_into ----------

func verifyDivideInto(ctx context.Context, legacy, target *sql.DB, opts Batch2Options,
	idx map[int64]string, rep *Batch2VerifyReport) error {
	type row struct {
		UserID               int64
		Type                 int32
		Quota, Income, Payed string
		Root, Parent         sql.NullInt64
		Left, Right, Layer   int32
	}
	const q = `
SELECT userid, cast(ifnull(type,1) as signed),
       CAST(ROUND(CAST(ifnull(quota,0) AS DECIMAL(20,3)),2) AS DECIMAL(12,2)),
       CAST(ROUND(CAST(ifnull(total_income,0) AS DECIMAL(20,3)),2) AS DECIMAL(12,2)),
       CAST(ROUND(CAST(ifnull(total_income_payed,0) AS DECIMAL(20,3)),2) AS DECIMAL(12,2)),
       root_userid, parent_userid,
       cast(ifnull(left_val,1) as signed), cast(ifnull(right_val,2) as signed),
       cast(ifnull(layer,1) as signed)
  FROM t_user_divide_into WHERE root_userid = ? ORDER BY left_val, userid`
	rows, err := legacy.QueryContext(ctx, q, opts.DivideRootUserID)
	if err != nil {
		return err
	}
	var exp []row
	for rows.Next() {
		var r row
		if err := rows.Scan(&r.UserID, &r.Type, &r.Quota, &r.Income, &r.Payed,
			&r.Root, &r.Parent, &r.Left, &r.Right, &r.Layer); err != nil {
			rows.Close()
			return err
		}
		exp = append(exp, r)
	}
	rows.Close()

	cUser := newCol("t_user_divide_into", "user_id", "legacy userid → ULID")
	cType := newCol("t_user_divide_into", "type", "")
	cQuota := newCol("t_user_divide_into", "quota", "float(10,2) → DECIMAL(12,2)")
	cIncome := newCol("t_user_divide_into", "total_income", "")
	cPayed := newCol("t_user_divide_into", "total_income_payed", "")
	cRoot := newCol("t_user_divide_into", "root_user_id", "根节点=自身")
	cParent := newCol("t_user_divide_into", "parent_user_id", "legacy parent_userid=0（根）→ NULL")
	cLeft := newCol("t_user_divide_into", "left_val", "Nested Set")
	cRight := newCol("t_user_divide_into", "right_val", "Nested Set")
	cLayer := newCol("t_user_divide_into", "layer", "")

	for _, e := range exp {
		rowKey := strconv.FormatInt(e.UserID, 10)
		wantUID := expectedULID(idx, sql.NullInt64{Int64: e.UserID, Valid: true})
		var got struct {
			UserID               string
			Type                 int8
			Quota, Income, Payed string
			Root, Parent         sql.NullString
			Left, Right, Layer   int32
		}
		err := target.QueryRowContext(ctx, `
SELECT user_id, type, CAST(quota AS DECIMAL(12,2)), CAST(total_income AS DECIMAL(12,2)),
       CAST(total_income_payed AS DECIMAL(12,2)), root_user_id, parent_user_id,
       left_val, right_val, layer
  FROM t_user_divide_into WHERE user_id = ?`, wantUID).Scan(
			&got.UserID, &got.Type, &got.Quota, &got.Income, &got.Payed,
			&got.Root, &got.Parent, &got.Left, &got.Right, &got.Layer)
		if err == sql.ErrNoRows {
			cUser.cmp(rowKey, wantUID, "<缺失>", false)
			continue
		}
		if err != nil {
			return err
		}
		cUser.cmp(rowKey, wantUID, got.UserID, got.UserID == wantUID)
		cType.cmp(rowKey, fmt.Sprint(e.Type), fmt.Sprint(got.Type), int32(got.Type) == e.Type)
		cQuota.cmp(rowKey, e.Quota, got.Quota, normalizeMoney(e.Quota) == normalizeMoney(got.Quota))
		cIncome.cmp(rowKey, e.Income, got.Income, normalizeMoney(e.Income) == normalizeMoney(got.Income))
		cPayed.cmp(rowKey, e.Payed, got.Payed, normalizeMoney(e.Payed) == normalizeMoney(got.Payed))
		cRoot.cmp(rowKey, expectedULID(idx, e.Root), nullStrOrNull(got.Root),
			nullStrOrNull(got.Root) == expectedULID(idx, e.Root))
		cParent.cmp(rowKey, expectedULID(idx, e.Parent), nullStrOrNull(got.Parent),
			nullStrOrNull(got.Parent) == expectedULID(idx, e.Parent))
		cLeft.cmp(rowKey, fmt.Sprint(e.Left), fmt.Sprint(got.Left), got.Left == e.Left)
		cRight.cmp(rowKey, fmt.Sprint(e.Right), fmt.Sprint(got.Right), got.Right == e.Right)
		cLayer.cmp(rowKey, fmt.Sprint(e.Layer), fmt.Sprint(got.Layer), got.Layer == e.Layer)
	}
	if len(exp) == 0 {
		cUser.cmp("-", "0 行", "0 行", false)
	}
	rep.Cols = append(rep.Cols, cUser.done(), cType.done(), cQuota.done(), cIncome.done(),
		cPayed.done(), cRoot.done(), cParent.done(), cLeft.done(), cRight.done(), cLayer.done())

	// 断言：树上**恰好 1 个根**（parent_user_id IS NULL），且该根的 root_user_id = user_id
	var total, nullParent, selfRoot int
	if err := target.QueryRowContext(ctx, `
SELECT COUNT(*),
       SUM(parent_user_id IS NULL),
       SUM(parent_user_id IS NULL AND root_user_id = user_id)
  FROM t_user_divide_into`).Scan(&total, &nullParent, &selfRoot); err != nil {
		return err
	}
	rep.Assertions = append(rep.Assertions, Assertion{
		Name: "分账树根节点 parent_user_id 为 NULL，且根节点 root_user_id = user_id（legacy parent_userid=0 / root_userid=自身）",
		Pass: nullParent == 1 && selfRoot == 1,
		Detail: fmt.Sprintf("t_user_divide_into 共 %d 行：parent_user_id IS NULL 的 %d 行（应=1），"+
			"其中 root_user_id=user_id 的 %d 行（应=1）", total, nullParent, selfRoot),
	})
	return nil
}

// ---------- t_user_divide_into_log ----------

func verifyDivideIntoLog(ctx context.Context, legacy, target *sql.DB, opts Batch2Options,
	idx map[int64]string, rep *Batch2VerifyReport) error {
	type row struct {
		ID      int64
		Type    int32
		UserID  int64
		Student int64
		OrderNo sql.NullString
		Money   string
		TimeStr sql.NullString
		Remark  sql.NullString
	}
	const q = `
SELECT id, cast(type as signed), userid, studentid, orderid,
       CAST(ROUND(CAST(ifnull(money,0) AS DECIMAL(20,3)),2) AS DECIMAL(12,2)),
       DATE_FORMAT(create_time,'%Y-%m-%d %H:%i:%s.%f'), remark
  FROM t_user_divide_into_log WHERE userid = ? ORDER BY id`
	rows, err := legacy.QueryContext(ctx, q, opts.DivideRootUserID)
	if err != nil {
		return err
	}
	var exp []row
	for rows.Next() {
		var r row
		if err := rows.Scan(&r.ID, &r.Type, &r.UserID, &r.Student, &r.OrderNo, &r.Money, &r.TimeStr, &r.Remark); err != nil {
			rows.Close()
			return err
		}
		exp = append(exp, r)
	}
	rows.Close()

	cID := newCol("t_user_divide_into_log", "id", "D1：与 legacy id 完全相同")
	cType := newCol("t_user_divide_into_log", "type", "tinyint zerofill → TINYINT")
	cUser := newCol("t_user_divide_into_log", "user_id", "userid → ULID")
	cStudent := newCol("t_user_divide_into_log", "student_id", "studentid → ULID")
	cOrder := newCol("t_user_divide_into_log", "order_no", "orderid → order_no")
	cMoney := newCol("t_user_divide_into_log", "money", "float(10,2) → DECIMAL(12,2)")
	cTime := newCol("t_user_divide_into_log", "create_time", "秒级相同，毫秒 .000")
	cRemark := newCol("t_user_divide_into_log", "remark", "text → LONGTEXT，报告脱敏")

	var legacyIDs, targetIDs []int64
	for _, e := range exp {
		rowKey := "id=" + strconv.FormatInt(e.ID, 10)
		legacyIDs = append(legacyIDs, e.ID)
		var g struct {
			ID      int64
			Type    int8
			UserID  string
			Student string
			OrderNo sql.NullString
			Money   string
			TimeStr sql.NullString
			Remark  sql.NullString
		}
		err := target.QueryRowContext(ctx, `
SELECT id, type, user_id, student_id, order_no,
       CAST(money AS DECIMAL(12,2)),
       DATE_FORMAT(create_time,'%Y-%m-%d %H:%i:%s.%f'), remark
  FROM t_user_divide_into_log WHERE id = ?`, e.ID).Scan(
			&g.ID, &g.Type, &g.UserID, &g.Student, &g.OrderNo, &g.Money, &g.TimeStr, &g.Remark)
		if err == sql.ErrNoRows {
			cID.cmp(rowKey, strconv.FormatInt(e.ID, 10), "<缺失>", false)
			continue
		}
		if err != nil {
			return err
		}
		targetIDs = append(targetIDs, g.ID)
		cID.cmp(rowKey, strconv.FormatInt(e.ID, 10), strconv.FormatInt(g.ID, 10), g.ID == e.ID)
		cType.cmp(rowKey, fmt.Sprint(e.Type), fmt.Sprint(g.Type), int32(g.Type) == e.Type)
		cUser.cmp(rowKey, expectedULID(idx, sql.NullInt64{Int64: e.UserID, Valid: true}), g.UserID,
			g.UserID == expectedULID(idx, sql.NullInt64{Int64: e.UserID, Valid: true}))
		cStudent.cmp(rowKey, expectedULID(idx, sql.NullInt64{Int64: e.Student, Valid: true}), g.Student,
			g.Student == expectedULID(idx, sql.NullInt64{Int64: e.Student, Valid: true}))
		cOrder.cmp(rowKey, nullStrOrNull(e.OrderNo), nullStrOrNull(g.OrderNo), nullStrOrNull(e.OrderNo) == nullStrOrNull(g.OrderNo))
		cMoney.cmp(rowKey, e.Money, g.Money, normalizeMoney(e.Money) == normalizeMoney(g.Money))
		cTime.cmp(rowKey, nullStrOrNull(e.TimeStr), nullStrOrNull(g.TimeStr), nullStrOrNull(e.TimeStr) == nullStrOrNull(g.TimeStr))
		cRemark.cmp(rowKey, maskRemark(e.Remark), maskRemark(g.Remark), sameRemark(e.Remark, g.Remark))
	}
	rep.Cols = append(rep.Cols, cID.done(), cType.done(), cUser.done(), cStudent.done(),
		cOrder.done(), cMoney.done(), cTime.done(), cRemark.done())

	rep.D1 = append(rep.D1, D1Verify{
		Table:     "t_user_divide_into_log",
		LegacyIDs: legacyIDs,
		TargetIDs: targetIDs,
		Pass:      int64SliceEqual(legacyIDs, targetIDs),
	})
	return nil
}

func int64SliceEqual(a, b []int64) bool {
	if len(a) != len(b) {
		return false
	}
	x := append([]int64(nil), a...)
	y := append([]int64(nil), b...)
	sort.Slice(x, func(i, j int) bool { return x[i] < x[j] })
	sort.Slice(y, func(i, j int) bool { return y[i] < y[j] })
	for i := range x {
		if x[i] != y[i] {
			return false
		}
	}
	return true
}

// ---------- t_user_balance_request ----------

func verifyBalanceRequest(ctx context.Context, legacy, target *sql.DB, opts Batch2Options,
	idx map[int64]string, rep *Batch2VerifyReport) error {
	type row struct {
		BizNo                       string
		Type                        int32
		UserID                      int64
		SchoolID, CampusID          int64
		Money                       sql.NullString
		TransferType                int8
		TransferTime                sql.NullString
		OrderNo                     sql.NullString
		ReqUser                     int64
		ReqTime                     sql.NullString
		RespUser                    sql.NullInt64
		RespTime                    sql.NullString
		RespMsg, Item, Imgs, Remark sql.NullString
		Aproved                     int8
	}
	const q = `
SELECT id, type, userid, schoolid, campusid,
       CAST(ROUND(CAST(money AS DECIMAL(20,3)),2) AS DECIMAL(12,2)),
       transfer_accounts_type, DATE_FORMAT(transfer_accounts_time,'%Y-%m-%d %H:%i:%s.%f'),
       orderid, request_userid, DATE_FORMAT(request_time,'%Y-%m-%d %H:%i:%s.%f'),
       response_userid, DATE_FORMAT(response_time,'%Y-%m-%d %H:%i:%s.%f'),
       response_msg, item, imgs, remark, aproved
  FROM t_user_balance_request WHERE userid = ? ORDER BY request_time, id`
	rows, err := legacy.QueryContext(ctx, q, opts.BalanceRequestUserID)
	if err != nil {
		return err
	}
	var exp []row
	for rows.Next() {
		var r row
		if err := rows.Scan(&r.BizNo, &r.Type, &r.UserID, &r.SchoolID, &r.CampusID, &r.Money,
			&r.TransferType, &r.TransferTime, &r.OrderNo, &r.ReqUser, &r.ReqTime,
			&r.RespUser, &r.RespTime, &r.RespMsg, &r.Item, &r.Imgs, &r.Remark, &r.Aproved); err != nil {
			rows.Close()
			return err
		}
		exp = append(exp, r)
	}
	rows.Close()

	const t = "t_user_balance_request"
	cBiz := newCol(t, "biz_no", "旧 UUID 主键原样保留（逐字符）")
	cType := newCol(t, "type", "")
	cUser := newCol(t, "user_id", "userid → ULID")
	cInst := newCol(t, "institution_id", "schoolid → institution_id（org 域未映射，直迁）")
	cCampus := newCol(t, "campus_id", "campusid → campus_id（同上）")
	cToInst := newCol(t, "to_institution_id", "legacy 无此列 → 必须 NULL")
	cToCampus := newCol(t, "to_campus_id", "legacy 无此列 → 必须 NULL")
	cMoney := newCol(t, "money", "decimal(10,3) → DECIMAL(12,2)")
	cTT := newCol(t, "transfer_accounts_type", "")
	cTTime := newCol(t, "transfer_accounts_time", "NULL → NULL")
	cOrder := newCol(t, "order_no", "orderid → order_no")
	cReqUser := newCol(t, "request_user_id", "request_userid → ULID")
	cReqTime := newCol(t, "request_time", "秒级相同")
	cRespUser := newCol(t, "response_user_id", "0 → NULL")
	cRespTime := newCol(t, "response_time", "NULL → NULL")
	cRespMsg := newCol(t, "response_msg", "空串保留空串")
	cApproved := newCol(t, "approved", "aproved 错拼修正；-2/1 原样")
	cItem := newCol(t, "item", "text → varchar(1000)，报告脱敏")
	cImgs := newCol(t, "imgs", "text → LONGTEXT，比对原文")
	cRemark := newCol(t, "remark", "空串保留空串，报告脱敏")

	var zeroResp int
	for _, e := range exp {
		rowKey := "biz=" + shortBizNo(e.BizNo)
		if e.RespUser.Valid && e.RespUser.Int64 == 0 {
			zeroResp++
		}
		var g struct {
			ID                                    int64
			BizNo                                 string
			Type                                  int32
			UserID                                string
			Inst, Campus                          int64
			ToInst, ToCampus                      sql.NullInt64
			Money                                 string
			TT                                    int8
			TTime, OrderNo, ReqTime               sql.NullString
			ReqUser                               string
			RespUser                              sql.NullString
			RespTime, RespMsg, Item, Imgs, Remark sql.NullString
			Approved                              int8
		}
		err := target.QueryRowContext(ctx, `
SELECT id, biz_no, type, user_id, institution_id, campus_id, to_institution_id, to_campus_id,
       CAST(money AS DECIMAL(12,2)), transfer_accounts_type,
       DATE_FORMAT(transfer_accounts_time,'%Y-%m-%d %H:%i:%s.%f'), order_no,
       request_user_id, DATE_FORMAT(request_time,'%Y-%m-%d %H:%i:%s.%f'),
       response_user_id, DATE_FORMAT(response_time,'%Y-%m-%d %H:%i:%s.%f'),
       response_msg, approved, item, imgs, remark
  FROM t_user_balance_request WHERE biz_no = ?`, e.BizNo).Scan(
			&g.ID, &g.BizNo, &g.Type, &g.UserID, &g.Inst, &g.Campus, &g.ToInst, &g.ToCampus,
			&g.Money, &g.TT, &g.TTime, &g.OrderNo, &g.ReqUser, &g.ReqTime, &g.RespUser,
			&g.RespTime, &g.RespMsg, &g.Approved, &g.Item, &g.Imgs, &g.Remark)
		if err == sql.ErrNoRows {
			cBiz.cmp(rowKey, e.BizNo, "<缺失>", false)
			continue
		}
		if err != nil {
			return err
		}
		cBiz.cmp(rowKey, e.BizNo, g.BizNo, g.BizNo == e.BizNo)
		cType.cmp(rowKey, fmt.Sprint(e.Type), fmt.Sprint(g.Type), g.Type == e.Type)
		cUser.cmp(rowKey, expectedULID(idx, sql.NullInt64{Int64: e.UserID, Valid: true}), g.UserID,
			g.UserID == expectedULID(idx, sql.NullInt64{Int64: e.UserID, Valid: true}))
		cInst.cmp(rowKey, fmt.Sprint(e.SchoolID), fmt.Sprint(g.Inst), g.Inst == e.SchoolID)
		cCampus.cmp(rowKey, fmt.Sprint(e.CampusID), fmt.Sprint(g.Campus), g.Campus == e.CampusID)
		cToInst.cmp(rowKey, "NULL", fmt.Sprint(g.ToInst.Int64), !g.ToInst.Valid)
		cToCampus.cmp(rowKey, "NULL", fmt.Sprint(g.ToCampus.Int64), !g.ToCampus.Valid)
		cMoney.cmp(rowKey, nullStrOrNull(e.Money), g.Money,
			normalizeMoney(orZero(e.Money)) == normalizeMoney(g.Money))
		cTT.cmp(rowKey, fmt.Sprint(e.TransferType), fmt.Sprint(g.TT), g.TT == e.TransferType)
		cTTime.cmp(rowKey, nullStrOrNull(e.TransferTime), nullStrOrNull(g.TTime),
			nullStrOrNull(e.TransferTime) == nullStrOrNull(g.TTime))
		cOrder.cmp(rowKey, nullStrOrNull(e.OrderNo), nullStrOrNull(g.OrderNo),
			nullStrOrNull(e.OrderNo) == nullStrOrNull(g.OrderNo))
		cReqUser.cmp(rowKey, expectedULID(idx, sql.NullInt64{Int64: e.ReqUser, Valid: true}), g.ReqUser,
			g.ReqUser == expectedULID(idx, sql.NullInt64{Int64: e.ReqUser, Valid: true}))
		cReqTime.cmp(rowKey, nullStrOrNull(e.ReqTime), nullStrOrNull(g.ReqTime),
			nullStrOrNull(e.ReqTime) == nullStrOrNull(g.ReqTime))
		cRespUser.cmp(rowKey, expectedULID(idx, e.RespUser), nullStrOrNull(g.RespUser),
			nullStrOrNull(g.RespUser) == expectedULID(idx, e.RespUser))
		cRespTime.cmp(rowKey, nullStrOrNull(e.RespTime), nullStrOrNull(g.RespTime),
			nullStrOrNull(e.RespTime) == nullStrOrNull(g.RespTime))
		cRespMsg.cmp(rowKey, maskText(e.RespMsg, 10), maskText(fromStr(g.RespMsg), 10), sameNullString(e.RespMsg, g.RespMsg))
		cApproved.cmp(rowKey, fmt.Sprint(e.Aproved), fmt.Sprint(g.Approved), g.Approved == e.Aproved)
		cItem.cmp(rowKey, maskText(e.Item, 16), maskText(fromStr(g.Item), 16), sameNullString(e.Item, g.Item))
		cImgs.cmp(rowKey, maskText(e.Imgs, 8), maskText(fromStr(g.Imgs), 8), sameNullString(e.Imgs, g.Imgs))
		cRemark.cmp(rowKey, maskText(e.Remark, 10), maskText(fromStr(g.Remark), 10), sameNullString(e.Remark, g.Remark))
	}
	rep.Cols = append(rep.Cols, cBiz.done(), cType.done(), cUser.done(), cInst.done(), cCampus.done(),
		cToInst.done(), cToCampus.done(), cMoney.done(), cTT.done(), cTTime.done(), cOrder.done(),
		cReqUser.done(), cReqTime.done(), cRespUser.done(), cRespTime.done(), cRespMsg.done(),
		cApproved.done(), cItem.done(), cImgs.done(), cRemark.done())

	rep.Assertions = append(rep.Assertions, Assertion{
		Name: "response_userid=0 → response_user_id 为 NULL（不是 0/空串）",
		Pass: countNonNullRespUser(ctx, target, opts.BalanceRequestUserID) == 0,
		Detail: fmt.Sprintf("legacy 有 %d 行 response_userid=0；target 侧该用户 response_user_id 非 NULL 且为空串的行数必须为 0",
			zeroResp),
	})
	return nil
}

// countNonNullRespUser 统计 fixture 用户下 response_user_id 落成 0/空串的行数（必须为 0）。
func countNonNullRespUser(ctx context.Context, db *sql.DB, legacyUserID int64) int {
	uid := FixtureCanonicalUserID(legacyUserID)
	var n int
	if err := db.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM t_user_balance_request
		  WHERE user_id = ? AND (response_user_id = '0' OR response_user_id = '')`, uid).Scan(&n); err != nil {
		return -1
	}
	return n
}

func orZero(v sql.NullString) string {
	if !v.Valid || v.String == "" {
		return "0.000"
	}
	return v.String
}

func fromStr(v sql.NullString) sql.NullString { return v }

func sameNullString(a, b sql.NullString) bool {
	if a.Valid != b.Valid {
		return false
	}
	if !a.Valid {
		return true
	}
	return a.String == b.String
}

// maskText 报告脱敏：只展示前 n 个字符 + 总长度。比对用原文（sameNullString）。
func maskText(v sql.NullString, n int) string {
	if !v.Valid {
		return "NULL"
	}
	r := []rune(v.String)
	head := r
	if len(head) > n {
		head = head[:n]
	}
	return fmt.Sprintf("%s…(len=%d)", string(head), len(r))
}

func shortBizNo(s string) string {
	if len(s) > 8 {
		return s[:8] + "…"
	}
	return s
}

// ---------- t_expenditure_log ----------

func verifyExpenditureLog(ctx context.Context, legacy, target *sql.DB, opts Batch2Options,
	idx map[int64]string, rep *Batch2VerifyReport) error {
	type row struct {
		BizNo              string
		SchoolID, CampusID int64
		ExpenditureID      int64
		Cost               string
		CreateUser         sql.NullInt64
		TimeStr            sql.NullString
		Year, Month        sql.NullInt64
		Remark, Imgs       sql.NullString
	}
	const q = `
SELECT id, schoolid, campusid, expenditureid,
       CAST(ROUND(CAST(cost AS DECIMAL(20,3)),2) AS DECIMAL(12,2)),
       create_userid, DATE_FORMAT(create_time,'%Y-%m-%d %H:%i:%s.%f'),
       year, month, remark, imgs
  FROM t_expenditure_log WHERE create_userid = ? ORDER BY id`
	rows, err := legacy.QueryContext(ctx, q, opts.ExpenditureLogUserID)
	if err != nil {
		return err
	}
	var exp []row
	for rows.Next() {
		var r row
		if err := rows.Scan(&r.BizNo, &r.SchoolID, &r.CampusID, &r.ExpenditureID, &r.Cost,
			&r.CreateUser, &r.TimeStr, &r.Year, &r.Month, &r.Remark, &r.Imgs); err != nil {
			rows.Close()
			return err
		}
		exp = append(exp, r)
	}
	rows.Close()

	const t = "t_expenditure_log"
	cBiz := newCol(t, "biz_no", "旧 UUID 主键原样保留（逐字符）")
	cInst := newCol(t, "institution_id", "schoolid → institution_id（org 域未映射，直迁）")
	cCampus := newCol(t, "campus_id", "campusid → campus_id（同上）")
	cExp := newCol(t, "expenditure_id", "expenditureid → t_expenditure.id（字典 D1，恒等）")
	cCost := newCol(t, "cost", "float(10,3) → DECIMAL(12,2)")
	cUser := newCol(t, "create_user_id", "create_userid → ULID；0 → NULL")
	cTime := newCol(t, "create_time", "秒级相同")
	cYear := newCol(t, "year", "MySQL YEAR → SMALLINT")
	cMonth := newCol(t, "month", "")
	cRemark := newCol(t, "remark", "text → LONGTEXT，报告脱敏")
	cImgs := newCol(t, "imgs", "text → LONGTEXT")

	for _, e := range exp {
		rowKey := "biz=" + shortBizNo(e.BizNo)
		var g struct {
			ID           int64
			BizNo        string
			Inst, Campus int64
			ExpID        int64
			Cost         string
			CreateUser   sql.NullString
			TimeStr      sql.NullString
			Year, Month  sql.NullInt64
			Remark, Imgs sql.NullString
		}
		err := target.QueryRowContext(ctx, `
SELECT id, biz_no, institution_id, campus_id, expenditure_id, CAST(cost AS DECIMAL(12,2)),
       create_user_id, DATE_FORMAT(create_time,'%Y-%m-%d %H:%i:%s.%f'),
       year, month, remark, imgs
  FROM t_expenditure_log WHERE biz_no = ?`, e.BizNo).Scan(
			&g.ID, &g.BizNo, &g.Inst, &g.Campus, &g.ExpID, &g.Cost, &g.CreateUser,
			&g.TimeStr, &g.Year, &g.Month, &g.Remark, &g.Imgs)
		if err == sql.ErrNoRows {
			cBiz.cmp(rowKey, e.BizNo, "<缺失>", false)
			continue
		}
		if err != nil {
			return err
		}
		cBiz.cmp(rowKey, e.BizNo, g.BizNo, g.BizNo == e.BizNo)
		cInst.cmp(rowKey, fmt.Sprint(e.SchoolID), fmt.Sprint(g.Inst), g.Inst == e.SchoolID)
		cCampus.cmp(rowKey, fmt.Sprint(e.CampusID), fmt.Sprint(g.Campus), g.Campus == e.CampusID)
		cExp.cmp(rowKey, fmt.Sprint(e.ExpenditureID), fmt.Sprint(g.ExpID), g.ExpID == e.ExpenditureID)
		cCost.cmp(rowKey, e.Cost, g.Cost, normalizeMoney(e.Cost) == normalizeMoney(g.Cost))
		cUser.cmp(rowKey, expectedULID(idx, e.CreateUser), nullStrOrNull(g.CreateUser),
			nullStrOrNull(g.CreateUser) == expectedULID(idx, e.CreateUser))
		cTime.cmp(rowKey, nullStrOrNull(e.TimeStr), nullStrOrNull(g.TimeStr),
			nullStrOrNull(e.TimeStr) == nullStrOrNull(g.TimeStr))
		cYear.cmp(rowKey, nullIntStr(e.Year), nullIntStr(g.Year), nullIntStr(e.Year) == nullIntStr(g.Year))
		cMonth.cmp(rowKey, nullIntStr(e.Month), nullIntStr(g.Month), nullIntStr(e.Month) == nullIntStr(g.Month))
		cRemark.cmp(rowKey, maskText(e.Remark, 10), maskText(g.Remark, 10), sameNullString(e.Remark, g.Remark))
		cImgs.cmp(rowKey, maskText(e.Imgs, 8), maskText(g.Imgs, 8), sameNullString(e.Imgs, g.Imgs))
	}
	rep.Cols = append(rep.Cols, cBiz.done(), cInst.done(), cCampus.done(), cExp.done(), cCost.done(),
		cUser.done(), cTime.done(), cYear.done(), cMonth.done(), cRemark.done(), cImgs.done())

	// 断言：expenditure_id 必须全部能在 t_expenditure 字典里找到
	var orphans int
	if err := target.QueryRowContext(ctx, `
SELECT COUNT(*) FROM t_expenditure_log l
  LEFT JOIN t_expenditure d ON d.id = l.expenditure_id
 WHERE d.id IS NULL`).Scan(&orphans); err != nil {
		return err
	}
	rep.Assertions = append(rep.Assertions, Assertion{
		Name:   "expenditure_id 全部可解析到 t_expenditure（无孤儿引用）",
		Pass:   orphans == 0,
		Detail: fmt.Sprintf("LEFT JOIN t_expenditure 后孤儿行数 = %d（字典 45 行走 D1，id 原样）", orphans),
	})
	return nil
}

func nullIntStr(v sql.NullInt64) string {
	if !v.Valid {
		return "NULL"
	}
	return strconv.FormatInt(v.Int64, 10)
}

// ---------- t_expenditure 字典 ----------

func verifyExpenditureDict(ctx context.Context, legacy, target *sql.DB, rep *Batch2VerifyReport) error {
	type row struct {
		ID       int64
		Type     int32
		ParentID int64
		Name     string
		Remark   sql.NullString
	}
	rows, err := legacy.QueryContext(ctx,
		`SELECT id, type, parentid, name, remark FROM t_expenditure ORDER BY id`)
	if err != nil {
		return err
	}
	var exp []row
	for rows.Next() {
		var r row
		if err := rows.Scan(&r.ID, &r.Type, &r.ParentID, &r.Name, &r.Remark); err != nil {
			rows.Close()
			return err
		}
		exp = append(exp, r)
	}
	rows.Close()

	const t = "t_expenditure"
	cID := newCol(t, "id", "D1：与 legacy id 完全相同")
	cType := newCol(t, "type", "")
	cParent := newCol(t, "parent_id", "自引用树，0=顶级")
	cName := newCol(t, "name", "字典名称原样")
	cRemark := newCol(t, "remark", "text → LONGTEXT")
	cInst := newCol(t, "institution_id", "legacy 无作用域列 → 0")

	for _, e := range exp {
		rowKey := "id=" + strconv.FormatInt(e.ID, 10)
		var g struct {
			ID, ParentID int64
			Type         int32
			Name         string
			Remark       sql.NullString
			Inst         int64
		}
		err := target.QueryRowContext(ctx,
			`SELECT id, type, parent_id, name, remark, institution_id FROM t_expenditure WHERE id = ?`,
			e.ID).Scan(&g.ID, &g.Type, &g.ParentID, &g.Name, &g.Remark, &g.Inst)
		if err == sql.ErrNoRows {
			cID.cmp(rowKey, fmt.Sprint(e.ID), "<缺失>", false)
			continue
		}
		if err != nil {
			return err
		}
		cID.cmp(rowKey, fmt.Sprint(e.ID), fmt.Sprint(g.ID), g.ID == e.ID)
		cType.cmp(rowKey, fmt.Sprint(e.Type), fmt.Sprint(g.Type), g.Type == e.Type)
		cParent.cmp(rowKey, fmt.Sprint(e.ParentID), fmt.Sprint(g.ParentID), g.ParentID == e.ParentID)
		cName.cmp(rowKey, e.Name, g.Name, g.Name == e.Name)
		cRemark.cmp(rowKey, maskText(e.Remark, 10), maskText(g.Remark, 10), sameNullString(e.Remark, g.Remark))
		cInst.cmp(rowKey, "0", fmt.Sprint(g.Inst), g.Inst == 0)
	}
	rep.Cols = append(rep.Cols, cID.done(), cType.done(), cParent.done(), cName.done(),
		cRemark.done(), cInst.done())
	return nil
}

// ---------- biz_no 四项验证 ----------

// verifyBizNoFour 逐条验证 biz_no 口径的四件事。
func verifyBizNoFour(ctx context.Context, legacy, target *sql.DB, opts Batch2Options,
	rep *Batch2VerifyReport) error {
	cases := []struct {
		table    string
		legacyQ  string
		arg      int64
		mirrorID func(bizNo string) string
	}{
		{
			table:   "t_user_balance_request",
			legacyQ: `SELECT id FROM t_user_balance_request WHERE userid = ? ORDER BY request_time, id`,
			arg:     opts.BalanceRequestUserID,
		},
		{
			table:   "t_expenditure_log",
			legacyQ: `SELECT id FROM t_expenditure_log WHERE create_userid = ? ORDER BY id`,
			arg:     opts.ExpenditureLogUserID,
		},
	}
	for _, c := range cases {
		v := BizNoVerify{Table: c.table}

		rows, err := legacy.QueryContext(ctx, c.legacyQ, c.arg)
		if err != nil {
			return err
		}
		var legacyBizNos []string
		for rows.Next() {
			var s string
			if err := rows.Scan(&s); err != nil {
				rows.Close()
				return err
			}
			legacyBizNos = append(legacyBizNos, s)
		}
		rows.Close()
		v.LegacyRows = len(legacyBizNos)

		// ① 逐字符相等 + ② 长度/hex + 唯一性
		trows, err := target.QueryContext(ctx, "SELECT biz_no, id FROM "+c.table)
		if err != nil {
			return err
		}
		targetByBiz := map[string]int64{}
		for trows.Next() {
			var b string
			var id int64
			if err := trows.Scan(&b, &id); err != nil {
				trows.Close()
				return err
			}
			targetByBiz[b] = id
		}
		trows.Close()
		v.Rows = len(targetByBiz)
		v.UniqueBizNo = len(targetByBiz)
		for _, b := range legacyBizNos {
			if _, ok := targetByBiz[b]; ok {
				v.CharEqual++
			}
			if len(b) == 32 {
				v.Len32++
			}
			if hex32.MatchString(b) {
				v.HexOK++
			}
		}

		// ③ id 与 biz_no 解耦：目标 id 是 1..N 的连续小整数，而 biz_no 是 32 位 hex；
		//    断言「id != biz_no 的任一数字派生」——这里用「id 必然 <= N」这一硬事实
		//    （若做过任何 hash/截断反算，id 会是大整数）。
		maxAllowed := int64(len(targetByBiz)) + 100
		derived := false
		for _, id := range targetByBiz {
			if id > maxAllowed {
				derived = true
			}
		}
		v.IDEqualsPrefix = derived // true = 疑似由 biz_no 反算，必须为 false

		// ④ 反查可用：用 biz_no JOIN 回 legacy
		var joined int
		if err := legacy.QueryRowContext(ctx, fmt.Sprintf(
			`SELECT COUNT(*) FROM %s r JOIN %s.%s t ON t.biz_no = r.id WHERE r.%s = ?`,
			c.table, "yuansiwei_2026", c.table, joinUserCol(c.table)), c.arg).Scan(&joined); err != nil {
			return err
		}
		v.ReverseJoin = joined

		// ⑤ id 解耦的**反序插入实验**（临时表，会话结束自动消失）：
		//    把同一批 biz_no **倒序**插进镜像表，若 id 跟着插入顺序涨、biz_no 保持原值，
		//    就说明 id 是纯插入计数器，与 biz_no 无关。
		mirrorOK, sample, err := mirrorBizNoOrder(ctx, target, c.table, legacyBizNos)
		if err != nil {
			return err
		}
		v.MirrorOK, v.MirrorSample = mirrorOK, sample

		// ⑥ AUTO_INCREMENT 未倒退
		ai, err := readAutoIncrement(ctx, target, c.table)
		if err != nil {
			return err
		}
		v.AutoIncrementVal = ai
		if err := target.QueryRowContext(ctx, "SELECT IFNULL(MAX(id),0) FROM "+c.table).Scan(&v.MaxID); err != nil {
			return err
		}
		rep.BizNo = append(rep.BizNo, v)
	}

	for _, v := range rep.BizNo {
		rep.Assertions = append(rep.Assertions,
			Assertion{
				Name: v.Table + ".biz_no 与 legacy id 逐字符相等",
				Pass: v.LegacyRows > 0 && v.CharEqual == v.LegacyRows && v.Rows == v.LegacyRows,
				Detail: fmt.Sprintf("legacy %d 行 / target %d 行 / 逐字符相等 %d 行；长度=32 的 %d 行，hex 形态 %d 行",
					v.LegacyRows, v.Rows, v.CharEqual, v.Len32, v.HexOK),
			},
			Assertion{
				Name:   v.Table + ".biz_no 唯一（uk_biz_no 生效，无重复）",
				Pass:   v.UniqueBizNo == v.Rows,
				Detail: fmt.Sprintf("DISTINCT biz_no = %d，总行数 = %d", v.UniqueBizNo, v.Rows),
			},
			Assertion{
				Name: v.Table + ".id 与 biz_no 解耦（id 是自增，不由 biz_no 派生）",
				Pass: !v.IDEqualsPrefix && v.MirrorOK,
				Detail: fmt.Sprintf("max(id)=%d，AUTO_INCREMENT=%d；反序插入镜像：%v",
					v.MaxID, v.AutoIncrementVal, v.MirrorSample),
			},
			Assertion{
				Name:   v.Table + ".biz_no 反查可用（JOIN 回 legacy 命中全部行）",
				Pass:   v.ReverseJoin == v.LegacyRows,
				Detail: fmt.Sprintf("JOIN finance_bk 命中 %d 行 / legacy %d 行", v.ReverseJoin, v.LegacyRows),
			},
			Assertion{
				Name:   v.Table + ".AUTO_INCREMENT >= max(id)+1（biz_no 口径下计数器正常递增）",
				Pass:   v.AutoIncrementVal >= v.MaxID+1,
				Detail: fmt.Sprintf("AUTO_INCREMENT=%d, max(id)=%d", v.AutoIncrementVal, v.MaxID),
			},
		)
	}
	return nil
}

// joinUserCol 返回各 biz_no 表在 JOIN 时用于筛选的列名。
func joinUserCol(table string) string {
	if table == "t_expenditure_log" {
		return "create_userid"
	}
	return "userid"
}

// mirrorRequiredCols 返回各 biz_no 表「NOT NULL 且无默认值」的列与哨兵值。
func mirrorRequiredCols(table string) ([]string, []any) {
	switch table {
	case "t_user_balance_request":
		// created_at / updated_at 在真实库里没有 DEFAULT（DDL 文件写了，实际库没生效）
		return []string{"created_at", "updated_at"}, []any{LegacyMissingTime, LegacyMissingTime}
	case "t_expenditure_log":
		return []string{"create_time"}, []any{LegacyMissingTime}
	default:
		return nil, nil
	}
}

// mirrorBizNoOrder 在临时表里**倒序**插入这批 biz_no，验证 id 只跟插入顺序、与 biz_no 无关。
func mirrorBizNoOrder(ctx context.Context, db *sql.DB, table string, bizNos []string) (bool, []string, error) {
	conn, err := db.Conn(ctx)
	if err != nil {
		return false, nil, err
	}
	defer conn.Close()

	tmp := "tmp_b2_bizno_" + table // 表名进临时表名：连接复用时避免残留旧结构
	// TEMPORARY TABLE 是**会话级**的：连接被复用时它还在，所以用 IF NOT EXISTS + 清空，
	// 全程不 DROP 任何真实表（DROP TEMPORARY 也只影响会话临时表，这里索性不用）。
	if _, err := conn.ExecContext(ctx,
		fmt.Sprintf("CREATE TEMPORARY TABLE IF NOT EXISTS `%s` LIKE %s", tmp, table)); err != nil {
		return false, nil, fmt.Errorf("建临时镜像表失败：%w", err)
	}
	if _, err := conn.ExecContext(ctx, fmt.Sprintf("DELETE FROM `%s`", tmp)); err != nil {
		return false, nil, fmt.Errorf("清空临时镜像表失败：%w", err)
	}
	if _, err := conn.ExecContext(ctx, fmt.Sprintf("ALTER TABLE `%s` AUTO_INCREMENT = 1", tmp)); err != nil {
		// 临时镜像表只用于观察发号起点；ALTER 只作用于会话临时表，不碰任何真实表。
		return false, nil, fmt.Errorf("重置临时镜像表自增失败：%w", err)
	}
	// 目标表的 NOT NULL 且无默认值的列必须显式给值（见报告「结构漂移」一节）。
	// 镜像表只用来观察 id 的发号规律，这些列填哨兵即可。
	extraCols, extraVals := mirrorRequiredCols(table)

	rev := append([]string(nil), bizNos...)
	sort.Sort(sort.Reverse(sort.StringSlice(rev)))
	for _, b := range rev {
		cols := "biz_no"
		holders := "?"
		args := []any{b}
		for i, c := range extraCols {
			cols += ", " + c
			holders += ", ?"
			args = append(args, extraVals[i])
		}
		if _, err := conn.ExecContext(ctx,
			fmt.Sprintf("INSERT INTO `%s` (%s) VALUES (%s)", tmp, cols, holders), args...); err != nil {
			return false, nil, fmt.Errorf("镜像插入失败：%w", err)
		}
	}
	rows, err := conn.QueryContext(ctx, fmt.Sprintf("SELECT id, biz_no FROM `%s` ORDER BY id", tmp))
	if err != nil {
		return false, nil, err
	}
	defer rows.Close()
	var sample []string
	var prevBiz string
	ok := true
	i := 0
	for rows.Next() {
		var id int64
		var b string
		if err := rows.Scan(&id, &b); err != nil {
			return false, nil, err
		}
		if len(sample) < 4 {
			sample = append(sample, fmt.Sprintf("id=%d←biz=%s", id, shortBizNo(b)))
		}
		// id 必须严格从 1 递增（插入顺序），而 biz_no 必须是**倒序**（与 id 顺序相反）
		if id != int64(i+1) {
			ok = false
		}
		if i > 0 && strings.Compare(b, prevBiz) >= 0 {
			ok = false
		}
		prevBiz = b
		i++
	}
	return ok && i == len(bizNos), sample, rows.Err()
}