| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256 |
- // 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"
- "yuansiwei/unified-backend/internal/infra/tz"
- )
- // 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=Asia%2FShanghai")。
- // - financeDsn string:教务库 DSN(如 "dev:pass@tcp(127.0.0.1:3306)/finance_bk?charset=utf8mb4&parseTime=True&loc=Asia%2FShanghai")。
- //
- // 出参:
- // - *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)
- }
- // 时区收口(与 API 进程同一口径):历史 datetime 一律按 Asia/Shanghai 解释,
- // 避免"迁移按 UTC 读、接口按 +08:00 写"造成同一条记录差 8 小时。
- teachingDsn = tz.NormalizeDSN(teachingDsn)
- financeDsn = tz.NormalizeDSN(financeDsn)
- 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
- }
|