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

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}

	// 落运行标记：create_time 补了 DEFAULT 之后，漏传业务时间会被 DB 静默填成插入时刻，
	// 只有 verify 侧「create_time >= 迁移执行时刻 的行数 == 0」这条断言能抓住它。
	if _, err := WriteRunMarker(ctx, target, "batch1"); err != nil {
		return nil, err
	}

	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

	// 运行标记也清掉（t_audit_log 应回到 0 行）
	if _, err := ClearRunMarkers(ctx, target); err != nil {
		return nil, err
	}

	// 复核 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) }