// Package migrate 数据迁移引擎(P1-04/T5,ADR-003 归一 pipeline 落地)。 // // 职责:从 legacy 双库(yuansiwei_i 教学 + finance_bk 教务,只读)读取三源用户数据, // 按 P0-06 归一规则 + P0-07 置信分级执行匹配,幂等 upsert 写入目标态 5 表 // (t_user/t_identity/t_wechat_union/t_membership/t_legacy_mapping)。 // // 设计铁律: // - **幂等**:重跑不产生重复行(按 legacy id 顺序处理 + 存在性检查) // - **先归一后匹配**:归一失败绝不写脏 canonical(phone_invalid → 仅记录 raw) // - **置信分级**:高置信(种子 + phone 唯一)自动写;中/低/冲突 → 仅出报告 // - **事务边界**:整批在单个 GORM 事务中(dry-run 不写) // - **ULID 发放**:id.NewCanonicalUserIDAt(regTime) 保证时间有序 // - **legacy 只读**:Source 连接仅执行 SELECT,无任何 INSERT/UPDATE/DELETE package migrate import ( "context" "fmt" "strings" "time" "gorm.io/driver/mysql" "gorm.io/gorm" "gorm.io/gorm/logger" ) // gormSilent 静默 logger(迁移批量读写高频 SQL,不刷日志;SLOW SQL 阈值噪音在此屏蔽)。 var gormSilent = logger.Default.LogMode(logger.Silent) // gormMigrateCfg 迁移包共用的 GORM 配置(不建物理 FK + 静默日志)。 func gormMigrateCfg() *gorm.Config { return &gorm.Config{ DisableForeignKeyConstraintWhenMigrating: true, Logger: gormSilent, } } // isMissingTable 判断 GORM 错误是否为"表不存在"(MySQL 1146)。 // dry-run / 幂等读路径据此容错:目标表未建时按"空表"处理(全量首次迁移)。 func isMissingTable(err error) bool { if err == nil { return false } msg := err.Error() return strings.Contains(msg, "1146") || strings.Contains(msg, "doesn't exist") } // ---- legacy 数据源结构体(GORM 只读映射,不 AutoMigrate)---- // LegacyUser 教学侧 yuansiwei_i.t_user(Canonical 主表,68107 行)。 type LegacyUser struct { ID int32 `gorm:"column:id;primaryKey"` Username string `gorm:"column:username"` RealName *string `gorm:"column:real_name"` Password string `gorm:"column:password"` Salt string `gorm:"column:salt"` Role int32 `gorm:"column:role"` Email *string `gorm:"column:email"` PhoneNumber *string `gorm:"column:phone_number"` Avatar *string `gorm:"column:avatar"` RegTime *time.Time `gorm:"column:reg_time"` Status int32 `gorm:"column:status"` ParentID int32 `gorm:"column:parent_id"` GradeID int32 `gorm:"column:grade_id"` } // TableName 指定 legacy 表名。 func (LegacyUser) TableName() string { return "t_user" } // LegacyWeixinUser 教学侧 yuansiwei_i.t_weixin_user(微信身份,3491 行)。 type LegacyWeixinUser struct { OpenID string `gorm:"column:openid;primaryKey"` Nickname string `gorm:"column:nickname"` UserID int32 `gorm:"column:userid"` UnionID *string `gorm:"column:unionid"` AppID *string `gorm:"column:appid"` } // TableName 指定 legacy 表名。 func (LegacyWeixinUser) TableName() string { return "t_weixin_user" } // LegacyAccountAssociation 教学侧 yuansiwei_i.t_user_account_association(现映射种子,16077 行)。 type LegacyAccountAssociation struct { UserID int32 `gorm:"column:userid;primaryKey"` MarkingSystemUserID *int32 `gorm:"column:marking_system_userid"` } // TableName 指定 legacy 表名。 func (LegacyAccountAssociation) TableName() string { return "t_user_account_association" } // LegacySchool 教学侧 yuansiwei_i.t_school(全国名录,仅取被引用的 23 所)。 type LegacySchool struct { ID int32 `gorm:"column:id;primaryKey"` Name string `gorm:"column:name"` Status int8 `gorm:"column:status"` // tinyint } // TableName 指定 legacy 表名。 func (LegacySchool) TableName() string { return "t_school" } // LegacySchoolCampus 教学侧 yuansiwei_i.t_school_campus(校区,63 行)。 type LegacySchoolCampus struct { ID int32 `gorm:"column:id;primaryKey"` Name string `gorm:"column:name"` SchoolID int32 `gorm:"column:school_id"` Status int8 `gorm:"column:status"` // tinyint Code *string `gorm:"column:code"` } // TableName 指定 legacy 表名。 func (LegacySchoolCampus) TableName() string { return "t_school_campus" } // LegacyUserSchoolCampus 教务侧 finance_bk.t_user_school_campus(用户-机构-校区绑定,26308 行)。 type LegacyUserSchoolCampus struct { ID int32 `gorm:"column:id;primaryKey"` UserID int32 `gorm:"column:userid"` SchoolID int32 `gorm:"column:schoolid"` CampusID int32 `gorm:"column:campusid"` } // TableName 指定 legacy 表名(finance_bk 库)。 func (LegacyUserSchoolCampus) TableName() string { return "t_user_school_campus" } // ---- 数据源接口 ---- // Source 封装 legacy 双库的只读 GORM 连接(教学 + 教务)。 // // 两个 *gorm.DB 分别指向 yuansiwei_i 和 finance_bk(同一 MySQL 实例,不同 database); // 若 DSN 未指定 database,用 UseDB 切换。 type Source struct { teaching *gorm.DB // yuansiwei_i(t_user/t_weixin_user/t_user_account_association/t_school/t_school_campus) finance *gorm.DB // finance_bk(t_user_school_campus) } // NewSource 创建 legacy 数据源(两个 DSN 指向同一 MySQL 实例的不同库)。 // // 入参: // - teachingDsn string:教学库 DSN(如 "dev:pass@tcp(127.0.0.1:3306)/yuansiwei_i?charset=utf8mb4&parseTime=True&loc=Local")。 // - financeDsn string:教务库 DSN(如 "dev:pass@tcp(127.0.0.1:3306)/finance_bk?charset=utf8mb4&parseTime=True&loc=Local")。 // // 出参: // - *Source:就绪的只读数据源。 // - error:连接失败时返回。 func NewSource(teachingDsn, financeDsn string) (*Source, error) { var missing []string if teachingDsn == "" { missing = append(missing, "teachingDsn") } if financeDsn == "" { missing = append(missing, "financeDsn") } if len(missing) > 0 { return nil, fmt.Errorf("migrate: legacy DSN 不能为空: %v", missing) } teaching, err := gorm.Open(mysql.Open(teachingDsn), gormMigrateCfg()) if err != nil { return nil, fmt.Errorf("migrate: 教学库连接失败: %w", err) } if sqlDB, err := teaching.DB(); err == nil { if err := sqlDB.Ping(); err != nil { _ = sqlDB.Close() return nil, fmt.Errorf("migrate: 教学库 Ping 失败: %w", err) } } finance, err := gorm.Open(mysql.Open(financeDsn), gormMigrateCfg()) if err != nil { return nil, fmt.Errorf("migrate: 教务库连接失败: %w", err) } if sqlDB, err := finance.DB(); err == nil { if err := sqlDB.Ping(); err != nil { _ = sqlDB.Close() return nil, fmt.Errorf("migrate: 教务库 Ping 失败: %w", err) } } return &Source{teaching: teaching, finance: finance}, nil } // Close 关闭两个连接池。 func (s *Source) Close() error { var firstErr error if s.teaching != nil { if sqlDB, err := s.teaching.DB(); err == nil { _ = sqlDB.Close() } } if s.finance != nil { if sqlDB, err := s.finance.DB(); err == nil { _ = sqlDB.Close() } else { firstErr = err } } return firstErr } // ---- 读取方法(全部 SELECT,无写操作)---- // LoadAllUsers 读取 t_user 全量(68107 行,按 id 排序保证幂等处理顺序)。 func (s *Source) LoadAllUsers(ctx context.Context) ([]LegacyUser, error) { var users []LegacyUser err := s.teaching.WithContext(ctx).Model(&LegacyUser{}).Order("id").Find(&users).Error return users, err } // LoadAllWeixinUsers 读取 t_weixin_user 全量(3491 行)。 func (s *Source) LoadAllWeixinUsers(ctx context.Context) ([]LegacyWeixinUser, error) { var rows []LegacyWeixinUser err := s.teaching.WithContext(ctx).Model(&LegacyWeixinUser{}).Order("openid").Find(&rows).Error return rows, err } // LoadAllAssociations 读取 t_user_account_association 全量(16077 行,现映射种子)。 func (s *Source) LoadAllAssociations(ctx context.Context) ([]LegacyAccountAssociation, error) { var rows []LegacyAccountAssociation err := s.teaching.WithContext(ctx).Model(&LegacyAccountAssociation{}).Order("userid").Find(&rows).Error return rows, err } // LoadSchools 读取 t_school 中被 t_user_school_campus 引用的学校(23 所)。 func (s *Source) LoadSchools(ctx context.Context, schoolIDs []int32) ([]LegacySchool, error) { if len(schoolIDs) == 0 { return nil, nil } var rows []LegacySchool err := s.teaching.WithContext(ctx).Model(&LegacySchool{}).Where("id IN ?", schoolIDs).Find(&rows).Error return rows, err } // LoadCampuses 读取 t_school_campus 全量(63 行)。 func (s *Source) LoadCampuses(ctx context.Context) ([]LegacySchoolCampus, error) { var rows []LegacySchoolCampus err := s.teaching.WithContext(ctx).Model(&LegacySchoolCampus{}).Order("id").Find(&rows).Error return rows, err } // LoadUserSchoolCampus 读取 finance_bk.t_user_school_campus 全量(26308 行)。 func (s *Source) LoadUserSchoolCampus(ctx context.Context) ([]LegacyUserSchoolCampus, error) { var rows []LegacyUserSchoolCampus err := s.finance.WithContext(ctx).Model(&LegacyUserSchoolCampus{}).Order("id").Find(&rows).Error return rows, err } // LoadDistinctSchoolIDs 从 t_user_school_campus 提取去重 schoolid(23 所)。 func (s *Source) LoadDistinctSchoolIDs(ctx context.Context) ([]int32, error) { var ids []int32 err := s.finance.WithContext(ctx).Model(&LegacyUserSchoolCampus{}). Distinct().Pluck("schoolid", &ids).Error return ids, err }