| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755 |
- 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),
- })
- }
- }
- 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 序,幂等)
- 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
- }
- 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, existUsername); 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, existPhone)
- r.writeWechatUnion(ctx, tx, u.ID, canon, planner, rep)
- if err := r.writeLegacyMapping(ctx, tx, u, canon, d, 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, 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
- }
- // ---- 写入方法(每个方法在 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)。
- func (r *Runner) writeUser(ctx context.Context, tx *gorm.DB, u *LegacyUser, canon string,
- d MatchDecision, rep *MigrateReport, usedUsername map[string]int32, existUsername map[string]int32) error {
- var n int64
- tx.WithContext(ctx).Model(&identitymodel.User{}).Where("canonical_user_id = ?", canon).Count(&n)
- if n > 0 {
- 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),
- })
- }
- }
- _ = existUsername // 存量 username 冲突已在 usedUsername 合并
- 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
- }
- rep.UsersCreated++
- if u.Status == 0 || u.Status == 3 {
- rep.UsersDisabled++
- }
- return nil
- }
- // writeIdentities 写 t_identity(phone/email/username/wechat;幂等 + 全局唯一键冲突处理)。
- func (r *Runner) writeIdentities(ctx context.Context, tx *gorm.DB, u *LegacyUser, canon string,
- d MatchDecision, rep *MigrateReport, usedPhone map[string]int32, existPhone map[string]int32) {
- // 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, "", 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, "", 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, "", 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, rep); err == nil {
- rep.IdentitiesWechat++
- }
- }
- }
- _ = existPhone
- }
- // wechatOf 取某 legacy user 的微信身份(runner 持有 planner 不方便,直接查 wx 缓存)。
- // 注:实现上在 Run 里通过闭包注入更合适;此处用字段缓存。
- func (r *Runner) wechatOf(userID int32) *LegacyWeixinUser {
- return r.wxCache[userID]
- }
- // upsertIdentity 幂等写入单条 identity(唯一键 type+identifier+appid 已存在则跳过)。
- func (r *Runner) upsertIdentity(ctx context.Context, tx *gorm.DB, canon, typ, identifier, appid string, rep *MigrateReport) error {
- var n int64
- tx.WithContext(ctx).Model(&identitymodel.Identity{}).
- Where("type = ? AND identifier = ? AND appid = ?", typ, identifier, appid).Count(&n)
- if n > 0 {
- rep.IdentityDuplicates++
- return nil
- }
- it := &identitymodel.Identity{
- CanonicalUserID: canon,
- Type: typ,
- Identifier: identifier,
- AppID: appid,
- Verified: false,
- CreatedAt: r.opts.Now,
- }
- return tx.WithContext(ctx).Create(it).Error
- }
- // writeWechatUnion 写 t_wechat_union(unionid 锚点;幂等:canonical/unionid 已存在则跳过)。
- func (r *Runner) writeWechatUnion(ctx context.Context, tx *gorm.DB, userID int32, canon string,
- planner *Planner, rep *MigrateReport) {
- w := r.wxCache[userID]
- if w == nil || w.UnionID == nil || *w.UnionID == "" {
- return
- }
- // 同 unionid 可能挂多个 canonical(跨 appid 同人):已存在任一 canonical 的该 unionid → 跳过
- var n int64
- tx.WithContext(ctx).Model(&identitymodel.WechatUnion{}).Where("unionid = ?", *w.UnionID).Count(&n)
- if n > 0 {
- 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 {
- 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 时天然存在则跳过)。
- func (r *Runner) writeLegacyMapping(ctx context.Context, tx *gorm.DB, u *LegacyUser, canon string,
- d MatchDecision, rep *MigrateReport) error {
- var n int64
- tx.WithContext(ctx).Model(&identitymodel.LegacyMapping{}).Where("canonical_user_id = ?", canon).Count(&n)
- if n > 0 {
- return nil // 已存在(重跑幂等)
- }
- legacyID := u.ID
- role := mapRoleCode(u.Role)
- lm := &identitymodel.LegacyMapping{
- CanonicalUserID: canon,
- LegacyIUserID: &legacyID, // I 库=教学侧,legacy_i_user_id
- InstitutionID: 0, // 平台级/待 P1-05 细化(canonical 用户跨机构)
- 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,
- }
- if err := tx.WithContext(ctx).Create(lm).Error; err != nil {
- return err
- }
- rep.LegacyMappingsCreated++
- return nil
- }
- // writeMemberships 写 t_membership(幂等:唯一键 canonical+campus+role)。
- 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, 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)
- var n int64
- tx.WithContext(ctx).Model(&orgmodel.Membership{}).
- Where("canonical_user_id = ? AND campus_id = ? AND role = ?", canon, campusID, role).Count(&n)
- if n > 0 {
- 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 {
- 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)
- }
|