DEV-MERGE-LEGACY-BACKEND-2026/internal/migrate/runner.go

package migrate

import (
	"context"
	"fmt"
	"time"

	"gorm.io/driver/mysql"
	"gorm.io/gorm"

	identitymodel "yuansiwei/unified-backend/internal/identity/model"
	"yuansiwei/unified-backend/internal/identity/id"
	"yuansiwei/unified-backend/internal/migrate/norm"
	orgmodel "yuansiwei/unified-backend/internal/org/model"
)

// Options 迁移运行配置。
type Options struct {
	// TeachingDSN 教学库 yuansiwei_i（只读源）。
	TeachingDSN string
	// FinanceDSN 教务库 finance_bk（只读源）。
	FinanceDSN string
	// TargetDSN 目标态 unified 库（读写，写入 5 表）。
	TargetDSN string
	// DryRun 仅预演：读库做存在性检查 + 产出报告，不执行任何写入。
	DryRun bool
	// Now 时间锚点（regTime 为空时用于 ULID 发放 / created_at）。可注入便于测试。
	Now time.Time
}

// Runner 迁移编排器（P1-04/T5 主体）。
//
// 职责：
//  1. 读 legacy 双库（只读）；
//  2. 按 P0-06 归一 + P0-07 置信分级规划；
//  3. 幂等 upsert 写目标态 5 表（t_institution/t_campus 维度 + t_user/t_identity/
//     t_wechat_union/t_membership/t_legacy_mapping）；
//  4. 产出 MigrateReport（含 P0-08 冲突/异常清单）。
//
// 事务边界：所有写入在单个 GORM 事务中（dry-run 不进入事务，仅读）。
type Runner struct {
	opts   Options
	src    *Source
	target *gorm.DB
	// wxCache：legacy userid → 微信身份（Run 时填充，writeIdentities/writeWechatUnion 消费）。
	// 放在 Runner 上避免大参数透传；非并发（单事务串行处理）。
	wxCache map[int32]*LegacyWeixinUser
}

// NewRunner 构建迁移编排器（连接 legacy 双库 + 目标库，均 fail-fast）。
func NewRunner(opts Options) (*Runner, error) {
	if opts.Now.IsZero() {
		opts.Now = time.Now()
	}
	src, err := NewSource(opts.TeachingDSN, opts.FinanceDSN)
	if err != nil {
		return nil, err
	}
	target, err := gorm.Open(gormMySQLTarget(opts.TargetDSN), gormMigrateCfg())
	if err != nil {
		_ = src.Close()
		return nil, fmt.Errorf("migrate: 目标库连接失败: %w", err)
	}
	if sqlDB, err := target.DB(); err == nil {
		if err := sqlDB.Ping(); err != nil {
			_ = src.Close()
			return nil, fmt.Errorf("migrate: 目标库 Ping 失败: %w", err)
		}
	}
	return &Runner{opts: opts, src: src, target: target}, nil
}

// Close 关闭所有连接。
func (r *Runner) Close() error {
	var err error
	if r.src != nil {
		err = r.src.Close()
	}
	if r.target != nil {
		if sqlDB, derr := r.target.DB(); derr == nil {
			_ = sqlDB.Close()
		}
	}
	return err
}

// EnsureTables 确保目标态 12 表存在（dev 期 AutoMigrate；生产以 DDL 为准，幂等对齐）。
// 迁移前调用，避免"写入时表不存在"。
func (r *Runner) EnsureTables(ctx context.Context) error {
	tables := append(identitymodel.AllTables(), orgmodel.AllTables()...)
	return r.target.WithContext(ctx).AutoMigrate(tables...)
}

// Run 执行迁移（单事务）。
//
// 出参：
//   - *MigrateReport：完整报告（含冲突/异常清单，P0-08 输入）。
//   - error：仅当读取/事务提交失败时返回；单行写入失败记录进 report.Warnings 不中断。
func (r *Runner) Run(ctx context.Context) (*MigrateReport, error) {
	rep := &MigrateReport{DryRun: r.opts.DryRun}

	// ---- 读取 legacy 全量（只读）----
	users, err := r.src.LoadAllUsers(ctx)
	if err != nil {
		return rep, fmt.Errorf("migrate: 读 t_user 失败: %w", err)
	}
	rep.UsersTotal = len(users)

	wx, err := r.src.LoadAllWeixinUsers(ctx)
	if err != nil {
		return rep, fmt.Errorf("migrate: 读 t_weixin_user 失败: %w", err)
	}
	// 填充微信缓存（userid → 身份，1:1）
	r.wxCache = make(map[int32]*LegacyWeixinUser, len(wx))
	for i := range wx {
		if wx[i].UserID > 0 {
			if _, exists := r.wxCache[wx[i].UserID]; !exists {
				r.wxCache[wx[i].UserID] = &wx[i]
			}
		}
	}
	assoc, err := r.src.LoadAllAssociations(ctx)
	if err != nil {
		return rep, fmt.Errorf("migrate: 读 t_user_account_association 失败: %w", err)
	}
	schoolIDs, err := r.src.LoadDistinctSchoolIDs(ctx)
	if err != nil {
		return rep, fmt.Errorf("migrate: 读 t_user_school_campus distinct schoolid 失败: %w", err)
	}
	schools, err := r.src.LoadSchools(ctx, schoolIDs)
	if err != nil {
		return rep, fmt.Errorf("migrate: 读 t_school 失败: %w", err)
	}
	rep.InstitutionsTotal = len(schools)
	campuses, err := r.src.LoadCampuses(ctx)
	if err != nil {
		return rep, fmt.Errorf("migrate: 读 t_school_campus 失败: %w", err)
	}
	rep.CampusesTotal = len(campuses)
	binds, err := r.src.LoadUserSchoolCampus(ctx)
	if err != nil {
		return rep, fmt.Errorf("migrate: 读 t_user_school_campus 失败: %w", err)
	}

	// ---- 规划（纯逻辑：ULID 发放 + 置信分级 + 冲突检测）----
	// 已有映射（重跑幂等：legacy id → canonical_user_id 复用）
	existingCanon, err := r.loadExistingMappings(ctx)
	if err != nil {
		return rep, fmt.Errorf("migrate: 读 t_legacy_mapping 失败: %w", err)
	}
	userSet := make(map[int32]struct{}, len(users))
	for i := range users {
		userSet[users[i].ID] = struct{}{}
	}
	// 微信孤儿（openid 的 userid 不在 t_user）
	for i := range wx {
		if _, ok := userSet[wx[i].UserID]; !ok {
			rep.WechatOrphans = append(rep.WechatOrphans, ConflictItem{
				LegacyID: wx[i].UserID,
				Detail:   fmt.Sprintf("openid=%s 的 userid 不在 t_user，丢弃", wx[i].OpenID),
			})
		}
	}

	// 种子映射（userid → 阅卷/标记系统 id），供 t_legacy_mapping.legacy_teaching_user_id（P1-05 §2 拍定）。
	teachingByUserID := make(map[int32]int32, len(assoc))
	for i := range assoc {
		if assoc[i].UserID > 0 && assoc[i].MarkingSystemUserID != nil && *assoc[i].MarkingSystemUserID > 0 {
			if _, exists := teachingByUserID[assoc[i].UserID]; !exists {
				teachingByUserID[assoc[i].UserID] = *assoc[i].MarkingSystemUserID
			}
		}
	}

	planner := NewPlanner(users, assoc, wx)
	// ULID 发放：已映射复用，新 id 发放（时间有序：以 regTime 为锚）
	canonByID := make(map[int32]string, len(users))
	for i := range users {
		u := &users[i]
		if c, ok := existingCanon[u.ID]; ok && c != "" {
			canonByID[u.ID] = c
			continue
		}
		var t time.Time
		if u.RegTime != nil {
			t = *u.RegTime
		} else {
			t = r.opts.Now
		}
		c, err := id.NewCanonicalUserIDAt(t)
		if err != nil {
			return rep, fmt.Errorf("migrate: ULID 发放失败(legacy id=%d): %w", u.ID, err)
		}
		canonByID[u.ID] = c
	}

	// 决策表（每个 user 的匹配决策）
	decisionByID := make(map[int32]MatchDecision, len(users))
	for i := range users {
		u := &users[i]
		d := planner.Decide(u.ID, u.PhoneNumber)
		decisionByID[u.ID] = d
		// phone 撞号冲突清单（每组只记一次，避免重复）
		if d.PhoneConflict {
			if u.PhoneNumber != nil {
				rep.PhoneConflicts = appendUniqueConflict(rep.PhoneConflicts, ConflictItem{
					LegacyID: u.ID,
					Detail:   fmt.Sprintf("phone=%s 撞号（同归一 phone 多账号，不自动合并）", normPhoneDisplay(*u.PhoneNumber)),
				})
			}
		}
		if d.PhoneInvalid && u.PhoneNumber != nil {
			rep.PhoneInvalid = append(rep.PhoneInvalid, ConflictItem{
				LegacyID: u.ID,
				Detail:   fmt.Sprintf("phone 归一失败，原始值=%q", *u.PhoneNumber),
			})
		}
	}

	// ---- 写入（单事务；dry-run 仅读不写）----
	if r.opts.DryRun {
		// dry-run：只做存在性预检 + 计数（不实际写入）。
		// 机构/校区/用户/身份/成员 的新建量按"目标库已有则复用"估算。
		r.simulateCounts(ctx, rep, schools, campuses, users, canonByID, binds)
		return rep, nil
	}

	err = r.target.Transaction(func(tx *gorm.DB) error {
		// ① 机构（维度表，幂等：已存在则复用）
		instByID, err := r.writeInstitutions(ctx, tx, schools, rep)
		if err != nil {
			return err
		}
		// ② 校区（依赖机构）
		campusByID, err := r.writeCampuses(ctx, tx, campuses, instByID, rep)
		if err != nil {
			return err
		}
		// ③ 用户 + 身份 + 微信锚点 + 遗留映射（按 legacy id 序，幂等）
		//
		// 性能：一次性批量装载目标库现有唯一键到内存 map（O(1) 查表替代 O(N) 逐行 COUNT）。
		// 插入时把新键标记为已占用，维持"首个占用者胜出"语义与重跑幂等。
		usedUsername := map[string]int32{} // 归一 username → 首个占用者
		usedPhone := map[string]int32{}   // 归一 phone → 首个占用者
		existUsername := r.existingUsernames(ctx, tx)
		existPhone := r.existingIdentities(ctx, tx, identitymodel.IdentityTypePhone, "")
		for k, v := range existUsername {
			usedUsername[k] = v
		}
		for k, v := range existPhone {
			usedPhone[k] = v
		}
		// 各表唯一键存量集合（批量装载，避免逐行 COUNT）
		existUserCanons := r.existingUserCanons(ctx, tx)      // t_user.canonical_user_id
		existIdentityKeys := r.existingIdentityKeys(ctx, tx)  // t_identity type|identifier|appid
		existUnionids := r.existingUnionids(ctx, tx)          // t_wechat_union.unionid
		existMappingCanons := r.existingMappingCanons(ctx, tx) // t_legacy_mapping.canonical_user_id
		existMemberKeys := r.existingMemberKeys(ctx, tx)      // t_membership canonical|campus|role

		for i := range users {
			u := &users[i]
			canon := canonByID[u.ID]
			d := decisionByID[u.ID]

			if err := r.writeUser(ctx, tx, u, canon, d, rep, usedUsername, existUserCanons); err != nil {
				rep.Warnings = append(rep.Warnings, fmt.Sprintf("写 t_user legacy_id=%d 失败: %v", u.ID, err))
				continue
			}
			r.writeIdentities(ctx, tx, u, canon, d, rep, usedPhone, existIdentityKeys)
			r.writeWechatUnion(ctx, tx, u.ID, canon, existUnionids, rep)
			if err := r.writeLegacyMapping(ctx, tx, u, canon, d, teachingByUserID, existMappingCanons, rep); err != nil {
				rep.Warnings = append(rep.Warnings, fmt.Sprintf("写 t_legacy_mapping legacy_id=%d 失败: %v", u.ID, err))
			}
		}
		// ④ 成员关系（依赖 canonical + 机构/校区）
		r.writeMemberships(ctx, tx, binds, canonByID, instByID, campusByID, users, existMemberKeys, rep)
		return nil
	})
	if err != nil {
		return rep, fmt.Errorf("migrate: 事务提交失败（已回滚）: %w", err)
	}
	return rep, nil
}

// normPhoneDisplay 展示用（截断，防超长）。
func normPhoneDisplay(raw string) string {
	c, ok := norm.Phone(raw)
	if ok {
		return c
	}
	return raw
}

// appendUniqueConflict 去重追加冲突项（同 LegacyID 只记一次）。
func appendUniqueConflict(items []ConflictItem, item ConflictItem) []ConflictItem {
	for _, e := range items {
		if e.LegacyID == item.LegacyID {
			return items
		}
	}
	return append(items, item)
}

// ---- 幂等辅助（读目标库现有状态）----

// loadExistingMappings 读 t_legacy_mapping，返回 legacy_i_user_id → canonical_user_id。
//
// dry-run 容错：目标表未建（MySQL 1146 表不存在）时按"空表"处理（全量首次迁移，ULID 全发放）。
// 这保证 dry-run 纯只读、不依赖目标库已建表。
func (r *Runner) loadExistingMappings(ctx context.Context) (map[int32]string, error) {
	m := make(map[int32]string)
	var rows []identitymodel.LegacyMapping
	if err := r.target.WithContext(ctx).Model(&identitymodel.LegacyMapping{}).Find(&rows).Error; err != nil {
		if isMissingTable(err) {
			return m, nil // 表不存在（dry-run 首次）→ 空映射
		}
		return nil, err
	}
	for i := range rows {
		if rows[i].LegacyIUserID != nil && *rows[i].LegacyIUserID > 0 {
			m[*rows[i].LegacyIUserID] = rows[i].CanonicalUserID
		}
	}
	return m, nil
}

// existingUsernames 读 t_user 现有 username（归一 username → canonical_user_id 占位，用于冲突检测）。
func (r *Runner) existingUsernames(ctx context.Context, tx *gorm.DB) map[string]int32 {
	m := make(map[string]int32)
	var users []identitymodel.User
	if err := tx.WithContext(ctx).Model(&identitymodel.User{}).Find(&users).Error; err != nil {
		return m
	}
	for i := range users {
		if users[i].Username != nil && *users[i].Username != "" {
			canon := norm.Username(*users[i].Username)
			if _, ok := m[canon]; !ok {
				m[canon] = 0 // 占位（无 legacy id，仅标记"已占用"）
			}
		}
	}
	return m
}

// existingIdentities 读 t_identity 现有指定 type 的 identifier（归一值 → 占位）。
func (r *Runner) existingIdentities(ctx context.Context, tx *gorm.DB, typ, appid string) map[string]int32 {
	m := make(map[string]int32)
	var idents []identitymodel.Identity
	q := tx.WithContext(ctx).Model(&identitymodel.Identity{}).Where("type = ?", typ)
	if err := q.Find(&idents).Error; err != nil {
		return m
	}
	for i := range idents {
		m[idents[i].Identifier] = 0
	}
	return m
}

// identityKey 唯一键 (type, identifier, appid) 的内存 map key（与 DB 唯一键 uk_type_identifier_appid 对齐）。
func identityKey(typ, identifier, appid string) string {
	return typ + "|" + identifier + "|" + appid
}

// memberKey 唯一键 (canonical_user_id, campus_id, role) 的内存 map key。
func memberKey(canon string, campus int64, role string) string {
	return canon + "|" + fmt.Sprint(campus) + "|" + role
}

// existingUserCanons 批量读 t_user.canonical_user_id → set（重跑幂等：已映射 ULID 复用，跳过写入）。
func (r *Runner) existingUserCanons(ctx context.Context, tx *gorm.DB) map[string]struct{} {
	m := make(map[string]struct{})
	var ids []string
	if err := tx.WithContext(ctx).Model(&identitymodel.User{}).
		Pluck("canonical_user_id", &ids).Error; err != nil {
		return m
	}
	for _, s := range ids {
		m[s] = struct{}{}
	}
	return m
}

// existingIdentityKeys 批量读 t_identity 全量唯一键 (type, identifier, appid) → set。
func (r *Runner) existingIdentityKeys(ctx context.Context, tx *gorm.DB) map[string]struct{} {
	m := make(map[string]struct{})
	var idents []identitymodel.Identity
	if err := tx.WithContext(ctx).Model(&identitymodel.Identity{}).Find(&idents).Error; err != nil {
		return m
	}
	for i := range idents {
		m[identityKey(idents[i].Type, idents[i].Identifier, idents[i].AppID)] = struct{}{}
	}
	return m
}

// existingUnionids 批量读 t_wechat_union.unionid → set。
func (r *Runner) existingUnionids(ctx context.Context, tx *gorm.DB) map[string]struct{} {
	m := make(map[string]struct{})
	var ids []string
	if err := tx.WithContext(ctx).Model(&identitymodel.WechatUnion{}).
		Pluck("unionid", &ids).Error; err != nil {
		return m
	}
	for _, s := range ids {
		m[s] = struct{}{}
	}
	return m
}

// existingMappingCanons 批量读 t_legacy_mapping.canonical_user_id → set。
func (r *Runner) existingMappingCanons(ctx context.Context, tx *gorm.DB) map[string]struct{} {
	m := make(map[string]struct{})
	var ids []string
	if err := tx.WithContext(ctx).Model(&identitymodel.LegacyMapping{}).
		Pluck("canonical_user_id", &ids).Error; err != nil {
		return m
	}
	for _, s := range ids {
		m[s] = struct{}{}
	}
	return m
}

// existingMemberKeys 批量读 t_membership 唯一键 (canonical_user_id, campus_id, role) → set。
func (r *Runner) existingMemberKeys(ctx context.Context, tx *gorm.DB) map[string]struct{} {
	m := make(map[string]struct{})
	var ms []orgmodel.Membership
	if err := tx.WithContext(ctx).Model(&orgmodel.Membership{}).
		Select("canonical_user_id", "campus_id", "role").Find(&ms).Error; err != nil {
		return m
	}
	for i := range ms {
		m[memberKey(ms[i].CanonicalUserID, ms[i].CampusID, ms[i].Role)] = struct{}{}
	}
	return m
}

// ---- 写入方法（每个方法在 dry-run 外执行；幂等：存在则跳过）----

// writeInstitutions 写 t_institution（幂等：按 name 查现有复用）。
func (r *Runner) writeInstitutions(ctx context.Context, tx *gorm.DB, schools []LegacySchool, rep *MigrateReport) (map[int32]int64, error) {
	instByID := make(map[int32]int64, len(schools))
	for i := range schools {
		s := &schools[i]
		var existing []orgmodel.Institution
		tx.WithContext(ctx).Model(&orgmodel.Institution{}).
			Where("institution_name = ?", s.Name).Limit(1).Find(&existing)
		if len(existing) > 0 {
			instByID[s.ID] = existing[0].InstitutionID
			continue
		}
		inst := &orgmodel.Institution{
			InstitutionName: s.Name,
			Status:          mapOrgStatus(s.Status),
			CreatedAt:       r.opts.Now,
			UpdatedAt:       r.opts.Now,
		}
		if err := tx.WithContext(ctx).Create(inst).Error; err != nil {
			return instByID, err
		}
		instByID[s.ID] = inst.InstitutionID
		rep.InstitutionsCreated++
	}
	return instByID, nil
}

// writeCampuses 写 t_campus（幂等：按 campus_name + institution 复用）。
func (r *Runner) writeCampuses(ctx context.Context, tx *gorm.DB, campuses []LegacySchoolCampus,
	instByID map[int32]int64, rep *MigrateReport) (map[int32]int64, error) {
	campusByID := make(map[int32]int64, len(campuses))
	for i := range campuses {
		c := &campuses[i]
		// 孤儿校区（school_id 不在被引用机构集）：institution_id 置 0（机构级/平台级）
		instID, ok := instByID[c.SchoolID]
		if !ok {
			instID = 0
		}
		var existing []orgmodel.Campus
		tx.WithContext(ctx).Model(&orgmodel.Campus{}).
			Where("campus_name = ? AND institution_id = ?", c.Name, instID).Limit(1).Find(&existing)
		if len(existing) > 0 {
			campusByID[c.ID] = existing[0].CampusID
			continue
		}
		campus := &orgmodel.Campus{
			InstitutionID: instID,
			CampusName:    c.Name,
			CampusCode:    c.Code,
			Status:        mapOrgStatus(c.Status),
			CreatedAt:     r.opts.Now,
			UpdatedAt:     r.opts.Now,
		}
		if err := tx.WithContext(ctx).Create(campus).Error; err != nil {
			return campusByID, err
		}
		campusByID[c.ID] = campus.CampusID
		rep.CampusesCreated++
	}
	return campusByID, nil
}

// writeUser 写 t_user（幂等：canonical_user_id 已存在则跳过；username 冲突置 NULL）。
//
// 性能：幂等判断用 existUserCanons 内存 set（事务内一次性批量装载），不再逐行 COUNT。
func (r *Runner) writeUser(ctx context.Context, tx *gorm.DB, u *LegacyUser, canon string,
	d MatchDecision, rep *MigrateReport, usedUsername map[string]int32, existUserCanons map[string]struct{}) error {
	if _, ok := existUserCanons[canon]; ok {
		rep.UsersSkipped++
		return nil
	}
	// username：归一后若已占用（存量或本批首个外）→ 置 NULL + 记冲突
	var username *string
	if u.Username != "" {
		canonUsername := norm.Username(u.Username)
		if _, taken := usedUsername[canonUsername]; !taken {
			s := u.Username
			username = &s
			usedUsername[canonUsername] = u.ID
		} else {
			rep.UsernameConflicts = append(rep.UsernameConflicts, ConflictItem{
				LegacyID: u.ID,
				Detail:   fmt.Sprintf("username=%q 归一后与已有用户重复，置 NULL", u.Username),
			})
		}
	}

	displayName := u.RealName
	user := &identitymodel.User{
		CanonicalUserID: canon,
		Username:        username,
		DisplayName:     displayName,
		AvatarURL:       u.Avatar,
		PersonType:      mapPersonType(u.Role),
		PasswordHash:    &u.Password,
		PasswordSalt:    &u.Salt,
		HashVersion:     strPtr(identitymodel.HashVersionShaSalt), // I 库=教学侧，sha_salt（ADR-016 惰性升级）
		Status:          mapUserStatus(u.Status),
		CreatedAt:       r.opts.Now,
		UpdatedAt:       r.opts.Now,
	}
	if u.RegTime != nil {
		user.CreatedAt = *u.RegTime
	}
	if err := tx.WithContext(ctx).Create(user).Error; err != nil {
		return err
	}
	existUserCanons[canon] = struct{}{} // 标记已占用（本批后续重复 ULID 幂等跳过）
	rep.UsersCreated++
	if u.Status == 0 || u.Status == 3 {
		rep.UsersDisabled++
	}
	return nil
}

// writeIdentities 写 t_identity（phone/email/username/wechat；幂等 + 全局唯一键冲突处理）。
//
// 性能：幂等判断用 existIdentityKeys 内存 set（事务内一次性批量装载），不再逐行 COUNT。
func (r *Runner) writeIdentities(ctx context.Context, tx *gorm.DB, u *LegacyUser, canon string,
	d MatchDecision, rep *MigrateReport, usedPhone map[string]int32, existIdentityKeys map[string]struct{}) {
	// phone（强归一；撞号→本批首个占用，其余跳过+冲突已记）
	if u.PhoneNumber != nil && *u.PhoneNumber != "" {
		canonPhone, ok := norm.Phone(*u.PhoneNumber)
		if ok && !d.PhoneConflict {
			if _, taken := usedPhone[canonPhone]; !taken {
				if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypePhone, canonPhone, "", existIdentityKeys, rep); err == nil {
					usedPhone[canonPhone] = u.ID
					rep.IdentitiesPhone++
				}
			}
		}
	}
	// email（轻归一）
	if u.Email != nil && *u.Email != "" {
		canonEmail, ok := norm.Email(*u.Email)
		if ok {
			if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeEmail, canonEmail, "", existIdentityKeys, rep); err == nil {
				rep.IdentitiesEmail++
			}
		}
	}
	// username（轻归一；与 t_user.username 一致）
	if u.Username != "" {
		canonUsername := norm.Username(u.Username)
		if canonUsername != "" {
			if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeUsername, canonUsername, "", existIdentityKeys, rep); err == nil {
				rep.IdentitiesUsername++
			}
		}
	}
	// wechat（openid + appid，经 t_weixin_user）
	if w := r.wechatOf(u.ID); w != nil {
		openid, ok := norm.WechatOpenID(w.OpenID)
		if ok {
			appid := ""
			if w.AppID != nil {
				appid = *w.AppID
			}
			if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeWechat, openid, appid, existIdentityKeys, rep); err == nil {
				rep.IdentitiesWechat++
			}
		}
	}
}

// wechatOf 取某 legacy user 的微信身份（runner 持有 planner 不方便，直接查 wx 缓存）。
// 注：实现上在 Run 里通过闭包注入更合适；此处用字段缓存。
func (r *Runner) wechatOf(userID int32) *LegacyWeixinUser {
	return r.wxCache[userID]
}

// upsertIdentity 幂等写入单条 identity（唯一键 type+identifier+appid 已存在则跳过）。
//
// 性能：幂等判断用 existIdentityKeys 内存 set（事务内一次性批量装载），不再逐行 COUNT。
func (r *Runner) upsertIdentity(ctx context.Context, tx *gorm.DB, canon, typ, identifier, appid string,
	existIdentityKeys map[string]struct{}, rep *MigrateReport) error {
	key := identityKey(typ, identifier, appid)
	if _, ok := existIdentityKeys[key]; ok {
		rep.IdentityDuplicates++
		return nil
	}
	it := &identitymodel.Identity{
		CanonicalUserID: canon,
		Type:            typ,
		Identifier:      identifier,
		AppID:           appid,
		Verified:        false,
		CreatedAt:       r.opts.Now,
	}
	if err := tx.WithContext(ctx).Create(it).Error; err != nil {
		return err
	}
	existIdentityKeys[key] = struct{}{} // 标记已占用
	return nil
}

// writeWechatUnion 写 t_wechat_union（unionid 锚点；幂等：unionid 已存在则跳过）。
//
// 性能：幂等判断用 existUnionids 内存 set（事务内一次性批量装载），不再逐行 COUNT。
func (r *Runner) writeWechatUnion(ctx context.Context, tx *gorm.DB, userID int32, canon string,
	existUnionids map[string]struct{}, rep *MigrateReport) {
	w := r.wxCache[userID]
	if w == nil || w.UnionID == nil || *w.UnionID == "" {
		return
	}
	// 同 unionid 可能挂多个 canonical（跨 appid 同人）：已存在该 unionid → 跳过
	if _, ok := existUnionids[*w.UnionID]; ok {
		return
	}
	wu := &identitymodel.WechatUnion{
		CanonicalUserID: canon,
		UnionID:         *w.UnionID,
		CreatedAt:       r.opts.Now,
		UpdatedAt:       r.opts.Now,
	}
	if err := tx.WithContext(ctx).Create(wu).Error; err == nil {
		existUnionids[*w.UnionID] = struct{}{} // 标记已占用
		rep.WechatUnionCreated++
	} else {
		rep.Warnings = append(rep.Warnings, fmt.Sprintf("写 t_wechat_union unionid=%s 失败: %v", *w.UnionID, err))
	}
}

// writeLegacyMapping 写 t_legacy_mapping（canonical_user_id UNIQUE 幂等；重跑复用 ULID 时天然存在则跳过）。
//
// 三源 user_id（P1-05 §2 拍定）：
//   - LegacyIUserID        = 旧A yuansiwei_i.t_user.id（本地主源，恒有）
//   - LegacyTeachingUserID = 阅卷/标记系统 id（t_user_account_association.marking_system_userid，仅种子用户有）
//   - LegacyAcademicUserID = 教务 user id（本地恒 NULL，finance_bk 无 user 主表）
func (r *Runner) writeLegacyMapping(ctx context.Context, tx *gorm.DB, u *LegacyUser, canon string,
	d MatchDecision, teachingByUserID map[int32]int32, existMappingCanons map[string]struct{}, rep *MigrateReport) error {
	if _, ok := existMappingCanons[canon]; ok {
		return nil // 已存在（重跑幂等 / 本批已写）
	}
	legacyID := u.ID
	role := mapRoleCode(u.Role)
	lm := &identitymodel.LegacyMapping{
		CanonicalUserID: canon,
		LegacyIUserID:   &legacyID, // 旧A yuansiwei_i.t_user.id
		InstitutionID:   0,         // 平台级（canonical 跨机构；代表机构非身份属性，P1-05 §6 拍定）
		CampusID:        0,
		Role:            &role,
		MatchSource:     strPtr(d.MatchSource),
		MatchConfidence: d.Confidence,
		AuditRef:        strPtr(fmt.Sprintf("migrate/v1.2/%d", u.ID)),
		MergedAt:        &r.opts.Now,
		CreatedAt:       r.opts.Now,
		UpdatedAt:       r.opts.Now,
	}
	// 阅卷/标记系统 id（种子用户才有；P1-05 §2：marking_system_userid → legacy_teaching_user_id）
	if markingID, ok := teachingByUserID[u.ID]; ok {
		lm.LegacyTeachingUserID = &markingID
	}
	if err := tx.WithContext(ctx).Create(lm).Error; err != nil {
		return err
	}
	existMappingCanons[canon] = struct{}{} // 标记已占用
	rep.LegacyMappingsCreated++
	return nil
}

// writeMemberships 写 t_membership（幂等：唯一键 canonical+campus+role）。
//
// 性能：幂等判断用 existMemberKeys 内存 set（事务内一次性批量装载），不再逐行 COUNT。
func (r *Runner) writeMemberships(ctx context.Context, tx *gorm.DB, binds []LegacyUserSchoolCampus,
	canonByID map[int32]string, instByID map[int32]int64, campusByID map[int32]int64,
	users []LegacyUser, existMemberKeys map[string]struct{}, rep *MigrateReport) {
	userByID := make(map[int32]*LegacyUser, len(users))
	for i := range users {
		userByID[users[i].ID] = &users[i]
	}
	for i := range binds {
		b := &binds[i]
		u, ok := userByID[b.UserID]
		if !ok {
			// 绑定用户不在 t_user（孤儿）→ 跳过 + 警告
			rep.Warnings = append(rep.Warnings, fmt.Sprintf("t_membership 跳过：legacy 绑定 userid=%d 不在 t_user", b.UserID))
			continue
		}
		canon, ok := canonByID[b.UserID]
		if !ok {
			continue
		}
		instID, ok1 := instByID[b.SchoolID]
		campusID, ok2 := campusByID[b.CampusID]
		if !ok1 || !ok2 {
			rep.Warnings = append(rep.Warnings, fmt.Sprintf("t_membership 跳过：institution/campus 映射缺失 schoolid=%d campusid=%d", b.SchoolID, b.CampusID))
			continue
		}
		role := mapRoleCode(u.Role)
		// 幂等：唯一键 (canonical, campus, role)
		if _, ok := existMemberKeys[memberKey(canon, campusID, role)]; ok {
			continue
		}
		m := &orgmodel.Membership{
			CanonicalUserID: canon,
			InstitutionID:   instID,
			CampusID:        campusID,
			Role:            role,
			Status:          orgmodel.MembershipStatusActive,
			JoinedAt:        &r.opts.Now,
			CreatedAt:       r.opts.Now,
		}
		if err := tx.WithContext(ctx).Create(m).Error; err == nil {
			existMemberKeys[memberKey(canon, campusID, role)] = struct{}{} // 标记已占用
			rep.MembershipsCreated++
		} else {
			rep.Warnings = append(rep.Warnings, fmt.Sprintf("写 t_membership legacy_userid=%d 失败: %v", b.UserID, err))
		}
	}
}

// simulateCounts dry-run 计数预检（只读，不写）：估算各表新建量。
func (r *Runner) simulateCounts(ctx context.Context, rep *MigrateReport, schools []LegacySchool,
	campuses []LegacySchoolCampus, users []LegacyUser, canonByID map[int32]string, binds []LegacyUserSchoolCampus) {
	existingUsers := r.countTable(ctx, "t_user")
	existingInst := r.countTable(ctx, "t_institution")
	existingCampus := r.countTable(ctx, "t_campus")
	existingMapping := r.countTable(ctx, "t_legacy_mapping")
	// 幂等估算：新建 = 目标量 - 已存在量（取下限 0）
	if len(users) > int(existingUsers) {
		rep.UsersCreated = len(users) - int(existingUsers)
	}
	if len(schools) > int(existingInst) {
		rep.InstitutionsCreated = len(schools) - int(existingInst)
	}
	if len(campuses) > int(existingCampus) {
		rep.CampusesCreated = len(campuses) - int(existingCampus)
	}
	if len(users) > int(existingMapping) {
		rep.LegacyMappingsCreated = len(users) - int(existingMapping)
	}
	// 身份/成员/锚点按"全量首次写入"估算（重跑时这些由唯一键去重，量趋 0）
	for i := range users {
		u := &users[i]
		_ = canonByID[u.ID]
		if u.PhoneNumber != nil && *u.PhoneNumber != "" {
			if _, ok := norm.Phone(*u.PhoneNumber); ok {
				rep.IdentitiesPhone++
			}
		}
		if u.Email != nil && *u.Email != "" {
			if _, ok := norm.Email(*u.Email); ok {
				rep.IdentitiesEmail++
			}
		}
		if u.Username != "" {
			rep.IdentitiesUsername++
		}
	}
	for i := range users {
		if w := r.wxCache[users[i].ID]; w != nil {
			rep.IdentitiesWechat++
			if w.UnionID != nil && *w.UnionID != "" {
				rep.WechatUnionCreated++
			}
		}
	}
	rep.MembershipsCreated = len(binds)
}

// countTable 统计表行数（dry-run 幂等估算用）。
func (r *Runner) countTable(ctx context.Context, table string) int64 {
	var n int64
	_ = r.target.WithContext(ctx).Table(table).Count(&n).Error
	return n
}

// ---- 映射函数 ----

// mapUserStatus legacy status int → 目标态 string。
// 1=active，0/2/3=disabled（P0-02 §4 拍板：0/3 都迁 disabled 保留可追溯）。
func mapUserStatus(legacy int32) string {
	if legacy == 1 {
		return identitymodel.UserStatusActive
	}
	return identitymodel.UserStatusDisabled
}

// mapOrgStatus 机构/校区 legacy status（tinyint）→ 目标态 string。
func mapOrgStatus(legacy int8) string {
	if legacy == 1 {
		return orgmodel.StatusActive
	}
	return orgmodel.StatusDisabled
}

// mapPersonType legacy role 数字码 → 目标态 person_type（P0-02 §3.1）。
func mapPersonType(legacy int32) string {
	switch legacy {
	case 2:
		return identitymodel.PersonTypeStudent
	case 4:
		return identitymodel.PersonTypeTeacher
	case 3:
		return identitymodel.PersonTypeParent
	default:
		return identitymodel.PersonTypeStaff
	}
}

// mapRoleCode legacy role 数字码 → 目标态 7 角色字符串码（P0-02 §2.5 拍板映射）。
func mapRoleCode(legacy int32) string {
	switch legacy {
	case 1:
		return orgmodel.RoleSuperAdmin
	case 2:
		return orgmodel.RoleStudent
	case 3:
		return orgmodel.RoleParent
	case 4:
		return orgmodel.RoleTeacher
	case 5, 6:
		return orgmodel.RoleAcademic // admin/redactor → 教务
	case 8:
		return orgmodel.RoleFinance
	case 7, 9:
		return orgmodel.RoleMarket // prototype/marketer → 市场
	default:
		return orgmodel.RoleAcademic // 未知角色兜底为教务（运营角色）
	}
}

// strPtr 字符串指针辅助。
func strPtr(s string) *string { return &s }

// gormMySQLTarget 目标库 GORM 驱动（mysql.Open 返回 Dialector，与 infra/db 同款）。
func gormMySQLTarget(dsn string) gorm.Dialector {
	return mysql.Open(dsn)
}