runner.go 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755
  1. package migrate
  2. import (
  3. "context"
  4. "fmt"
  5. "time"
  6. "gorm.io/driver/mysql"
  7. "gorm.io/gorm"
  8. identitymodel "yuansiwei/unified-backend/internal/identity/model"
  9. "yuansiwei/unified-backend/internal/identity/id"
  10. "yuansiwei/unified-backend/internal/migrate/norm"
  11. orgmodel "yuansiwei/unified-backend/internal/org/model"
  12. )
  13. // Options 迁移运行配置。
  14. type Options struct {
  15. // TeachingDSN 教学库 yuansiwei_i(只读源)。
  16. TeachingDSN string
  17. // FinanceDSN 教务库 finance_bk(只读源)。
  18. FinanceDSN string
  19. // TargetDSN 目标态 unified 库(读写,写入 5 表)。
  20. TargetDSN string
  21. // DryRun 仅预演:读库做存在性检查 + 产出报告,不执行任何写入。
  22. DryRun bool
  23. // Now 时间锚点(regTime 为空时用于 ULID 发放 / created_at)。可注入便于测试。
  24. Now time.Time
  25. }
  26. // Runner 迁移编排器(P1-04/T5 主体)。
  27. //
  28. // 职责:
  29. // 1. 读 legacy 双库(只读);
  30. // 2. 按 P0-06 归一 + P0-07 置信分级规划;
  31. // 3. 幂等 upsert 写目标态 5 表(t_institution/t_campus 维度 + t_user/t_identity/
  32. // t_wechat_union/t_membership/t_legacy_mapping);
  33. // 4. 产出 MigrateReport(含 P0-08 冲突/异常清单)。
  34. //
  35. // 事务边界:所有写入在单个 GORM 事务中(dry-run 不进入事务,仅读)。
  36. type Runner struct {
  37. opts Options
  38. src *Source
  39. target *gorm.DB
  40. // wxCache:legacy userid → 微信身份(Run 时填充,writeIdentities/writeWechatUnion 消费)。
  41. // 放在 Runner 上避免大参数透传;非并发(单事务串行处理)。
  42. wxCache map[int32]*LegacyWeixinUser
  43. }
  44. // NewRunner 构建迁移编排器(连接 legacy 双库 + 目标库,均 fail-fast)。
  45. func NewRunner(opts Options) (*Runner, error) {
  46. if opts.Now.IsZero() {
  47. opts.Now = time.Now()
  48. }
  49. src, err := NewSource(opts.TeachingDSN, opts.FinanceDSN)
  50. if err != nil {
  51. return nil, err
  52. }
  53. target, err := gorm.Open(gormMySQLTarget(opts.TargetDSN), gormMigrateCfg())
  54. if err != nil {
  55. _ = src.Close()
  56. return nil, fmt.Errorf("migrate: 目标库连接失败: %w", err)
  57. }
  58. if sqlDB, err := target.DB(); err == nil {
  59. if err := sqlDB.Ping(); err != nil {
  60. _ = src.Close()
  61. return nil, fmt.Errorf("migrate: 目标库 Ping 失败: %w", err)
  62. }
  63. }
  64. return &Runner{opts: opts, src: src, target: target}, nil
  65. }
  66. // Close 关闭所有连接。
  67. func (r *Runner) Close() error {
  68. var err error
  69. if r.src != nil {
  70. err = r.src.Close()
  71. }
  72. if r.target != nil {
  73. if sqlDB, derr := r.target.DB(); derr == nil {
  74. _ = sqlDB.Close()
  75. }
  76. }
  77. return err
  78. }
  79. // EnsureTables 确保目标态 12 表存在(dev 期 AutoMigrate;生产以 DDL 为准,幂等对齐)。
  80. // 迁移前调用,避免"写入时表不存在"。
  81. func (r *Runner) EnsureTables(ctx context.Context) error {
  82. tables := append(identitymodel.AllTables(), orgmodel.AllTables()...)
  83. return r.target.WithContext(ctx).AutoMigrate(tables...)
  84. }
  85. // Run 执行迁移(单事务)。
  86. //
  87. // 出参:
  88. // - *MigrateReport:完整报告(含冲突/异常清单,P0-08 输入)。
  89. // - error:仅当读取/事务提交失败时返回;单行写入失败记录进 report.Warnings 不中断。
  90. func (r *Runner) Run(ctx context.Context) (*MigrateReport, error) {
  91. rep := &MigrateReport{DryRun: r.opts.DryRun}
  92. // ---- 读取 legacy 全量(只读)----
  93. users, err := r.src.LoadAllUsers(ctx)
  94. if err != nil {
  95. return rep, fmt.Errorf("migrate: 读 t_user 失败: %w", err)
  96. }
  97. rep.UsersTotal = len(users)
  98. wx, err := r.src.LoadAllWeixinUsers(ctx)
  99. if err != nil {
  100. return rep, fmt.Errorf("migrate: 读 t_weixin_user 失败: %w", err)
  101. }
  102. // 填充微信缓存(userid → 身份,1:1)
  103. r.wxCache = make(map[int32]*LegacyWeixinUser, len(wx))
  104. for i := range wx {
  105. if wx[i].UserID > 0 {
  106. if _, exists := r.wxCache[wx[i].UserID]; !exists {
  107. r.wxCache[wx[i].UserID] = &wx[i]
  108. }
  109. }
  110. }
  111. assoc, err := r.src.LoadAllAssociations(ctx)
  112. if err != nil {
  113. return rep, fmt.Errorf("migrate: 读 t_user_account_association 失败: %w", err)
  114. }
  115. schoolIDs, err := r.src.LoadDistinctSchoolIDs(ctx)
  116. if err != nil {
  117. return rep, fmt.Errorf("migrate: 读 t_user_school_campus distinct schoolid 失败: %w", err)
  118. }
  119. schools, err := r.src.LoadSchools(ctx, schoolIDs)
  120. if err != nil {
  121. return rep, fmt.Errorf("migrate: 读 t_school 失败: %w", err)
  122. }
  123. rep.InstitutionsTotal = len(schools)
  124. campuses, err := r.src.LoadCampuses(ctx)
  125. if err != nil {
  126. return rep, fmt.Errorf("migrate: 读 t_school_campus 失败: %w", err)
  127. }
  128. rep.CampusesTotal = len(campuses)
  129. binds, err := r.src.LoadUserSchoolCampus(ctx)
  130. if err != nil {
  131. return rep, fmt.Errorf("migrate: 读 t_user_school_campus 失败: %w", err)
  132. }
  133. // ---- 规划(纯逻辑:ULID 发放 + 置信分级 + 冲突检测)----
  134. // 已有映射(重跑幂等:legacy id → canonical_user_id 复用)
  135. existingCanon, err := r.loadExistingMappings(ctx)
  136. if err != nil {
  137. return rep, fmt.Errorf("migrate: 读 t_legacy_mapping 失败: %w", err)
  138. }
  139. userSet := make(map[int32]struct{}, len(users))
  140. for i := range users {
  141. userSet[users[i].ID] = struct{}{}
  142. }
  143. // 微信孤儿(openid 的 userid 不在 t_user)
  144. for i := range wx {
  145. if _, ok := userSet[wx[i].UserID]; !ok {
  146. rep.WechatOrphans = append(rep.WechatOrphans, ConflictItem{
  147. LegacyID: wx[i].UserID,
  148. Detail: fmt.Sprintf("openid=%s 的 userid 不在 t_user,丢弃", wx[i].OpenID),
  149. })
  150. }
  151. }
  152. planner := NewPlanner(users, assoc, wx)
  153. // ULID 发放:已映射复用,新 id 发放(时间有序:以 regTime 为锚)
  154. canonByID := make(map[int32]string, len(users))
  155. for i := range users {
  156. u := &users[i]
  157. if c, ok := existingCanon[u.ID]; ok && c != "" {
  158. canonByID[u.ID] = c
  159. continue
  160. }
  161. var t time.Time
  162. if u.RegTime != nil {
  163. t = *u.RegTime
  164. } else {
  165. t = r.opts.Now
  166. }
  167. c, err := id.NewCanonicalUserIDAt(t)
  168. if err != nil {
  169. return rep, fmt.Errorf("migrate: ULID 发放失败(legacy id=%d): %w", u.ID, err)
  170. }
  171. canonByID[u.ID] = c
  172. }
  173. // 决策表(每个 user 的匹配决策)
  174. decisionByID := make(map[int32]MatchDecision, len(users))
  175. for i := range users {
  176. u := &users[i]
  177. d := planner.Decide(u.ID, u.PhoneNumber)
  178. decisionByID[u.ID] = d
  179. // phone 撞号冲突清单(每组只记一次,避免重复)
  180. if d.PhoneConflict {
  181. if u.PhoneNumber != nil {
  182. rep.PhoneConflicts = appendUniqueConflict(rep.PhoneConflicts, ConflictItem{
  183. LegacyID: u.ID,
  184. Detail: fmt.Sprintf("phone=%s 撞号(同归一 phone 多账号,不自动合并)", normPhoneDisplay(*u.PhoneNumber)),
  185. })
  186. }
  187. }
  188. if d.PhoneInvalid && u.PhoneNumber != nil {
  189. rep.PhoneInvalid = append(rep.PhoneInvalid, ConflictItem{
  190. LegacyID: u.ID,
  191. Detail: fmt.Sprintf("phone 归一失败,原始值=%q", *u.PhoneNumber),
  192. })
  193. }
  194. }
  195. // ---- 写入(单事务;dry-run 仅读不写)----
  196. if r.opts.DryRun {
  197. // dry-run:只做存在性预检 + 计数(不实际写入)。
  198. // 机构/校区/用户/身份/成员 的新建量按"目标库已有则复用"估算。
  199. r.simulateCounts(ctx, rep, schools, campuses, users, canonByID, binds)
  200. return rep, nil
  201. }
  202. err = r.target.Transaction(func(tx *gorm.DB) error {
  203. // ① 机构(维度表,幂等:已存在则复用)
  204. instByID, err := r.writeInstitutions(ctx, tx, schools, rep)
  205. if err != nil {
  206. return err
  207. }
  208. // ② 校区(依赖机构)
  209. campusByID, err := r.writeCampuses(ctx, tx, campuses, instByID, rep)
  210. if err != nil {
  211. return err
  212. }
  213. // ③ 用户 + 身份 + 微信锚点 + 遗留映射(按 legacy id 序,幂等)
  214. usedUsername := map[string]int32{} // 归一 username → 首个占用者
  215. usedPhone := map[string]int32{} // 归一 phone → 首个占用者
  216. existUsername := r.existingUsernames(ctx, tx)
  217. existPhone := r.existingIdentities(ctx, tx, identitymodel.IdentityTypePhone, "")
  218. for k, v := range existUsername {
  219. usedUsername[k] = v
  220. }
  221. for k, v := range existPhone {
  222. usedPhone[k] = v
  223. }
  224. for i := range users {
  225. u := &users[i]
  226. canon := canonByID[u.ID]
  227. d := decisionByID[u.ID]
  228. if err := r.writeUser(ctx, tx, u, canon, d, rep, usedUsername, existUsername); err != nil {
  229. rep.Warnings = append(rep.Warnings, fmt.Sprintf("写 t_user legacy_id=%d 失败: %v", u.ID, err))
  230. continue
  231. }
  232. r.writeIdentities(ctx, tx, u, canon, d, rep, usedPhone, existPhone)
  233. r.writeWechatUnion(ctx, tx, u.ID, canon, planner, rep)
  234. if err := r.writeLegacyMapping(ctx, tx, u, canon, d, rep); err != nil {
  235. rep.Warnings = append(rep.Warnings, fmt.Sprintf("写 t_legacy_mapping legacy_id=%d 失败: %v", u.ID, err))
  236. }
  237. }
  238. // ④ 成员关系(依赖 canonical + 机构/校区)
  239. r.writeMemberships(ctx, tx, binds, canonByID, instByID, campusByID, users, rep)
  240. return nil
  241. })
  242. if err != nil {
  243. return rep, fmt.Errorf("migrate: 事务提交失败(已回滚): %w", err)
  244. }
  245. return rep, nil
  246. }
  247. // normPhoneDisplay 展示用(截断,防超长)。
  248. func normPhoneDisplay(raw string) string {
  249. c, ok := norm.Phone(raw)
  250. if ok {
  251. return c
  252. }
  253. return raw
  254. }
  255. // appendUniqueConflict 去重追加冲突项(同 LegacyID 只记一次)。
  256. func appendUniqueConflict(items []ConflictItem, item ConflictItem) []ConflictItem {
  257. for _, e := range items {
  258. if e.LegacyID == item.LegacyID {
  259. return items
  260. }
  261. }
  262. return append(items, item)
  263. }
  264. // ---- 幂等辅助(读目标库现有状态)----
  265. // loadExistingMappings 读 t_legacy_mapping,返回 legacy_i_user_id → canonical_user_id。
  266. //
  267. // dry-run 容错:目标表未建(MySQL 1146 表不存在)时按"空表"处理(全量首次迁移,ULID 全发放)。
  268. // 这保证 dry-run 纯只读、不依赖目标库已建表。
  269. func (r *Runner) loadExistingMappings(ctx context.Context) (map[int32]string, error) {
  270. m := make(map[int32]string)
  271. var rows []identitymodel.LegacyMapping
  272. if err := r.target.WithContext(ctx).Model(&identitymodel.LegacyMapping{}).Find(&rows).Error; err != nil {
  273. if isMissingTable(err) {
  274. return m, nil // 表不存在(dry-run 首次)→ 空映射
  275. }
  276. return nil, err
  277. }
  278. for i := range rows {
  279. if rows[i].LegacyIUserID != nil && *rows[i].LegacyIUserID > 0 {
  280. m[*rows[i].LegacyIUserID] = rows[i].CanonicalUserID
  281. }
  282. }
  283. return m, nil
  284. }
  285. // existingUsernames 读 t_user 现有 username(归一 username → canonical_user_id 占位,用于冲突检测)。
  286. func (r *Runner) existingUsernames(ctx context.Context, tx *gorm.DB) map[string]int32 {
  287. m := make(map[string]int32)
  288. var users []identitymodel.User
  289. if err := tx.WithContext(ctx).Model(&identitymodel.User{}).Find(&users).Error; err != nil {
  290. return m
  291. }
  292. for i := range users {
  293. if users[i].Username != nil && *users[i].Username != "" {
  294. canon := norm.Username(*users[i].Username)
  295. if _, ok := m[canon]; !ok {
  296. m[canon] = 0 // 占位(无 legacy id,仅标记"已占用")
  297. }
  298. }
  299. }
  300. return m
  301. }
  302. // existingIdentities 读 t_identity 现有指定 type 的 identifier(归一值 → 占位)。
  303. func (r *Runner) existingIdentities(ctx context.Context, tx *gorm.DB, typ, appid string) map[string]int32 {
  304. m := make(map[string]int32)
  305. var idents []identitymodel.Identity
  306. q := tx.WithContext(ctx).Model(&identitymodel.Identity{}).Where("type = ?", typ)
  307. if err := q.Find(&idents).Error; err != nil {
  308. return m
  309. }
  310. for i := range idents {
  311. m[idents[i].Identifier] = 0
  312. }
  313. return m
  314. }
  315. // ---- 写入方法(每个方法在 dry-run 外执行;幂等:存在则跳过)----
  316. // writeInstitutions 写 t_institution(幂等:按 name 查现有复用)。
  317. func (r *Runner) writeInstitutions(ctx context.Context, tx *gorm.DB, schools []LegacySchool, rep *MigrateReport) (map[int32]int64, error) {
  318. instByID := make(map[int32]int64, len(schools))
  319. for i := range schools {
  320. s := &schools[i]
  321. var existing []orgmodel.Institution
  322. tx.WithContext(ctx).Model(&orgmodel.Institution{}).
  323. Where("institution_name = ?", s.Name).Limit(1).Find(&existing)
  324. if len(existing) > 0 {
  325. instByID[s.ID] = existing[0].InstitutionID
  326. continue
  327. }
  328. inst := &orgmodel.Institution{
  329. InstitutionName: s.Name,
  330. Status: mapOrgStatus(s.Status),
  331. CreatedAt: r.opts.Now,
  332. UpdatedAt: r.opts.Now,
  333. }
  334. if err := tx.WithContext(ctx).Create(inst).Error; err != nil {
  335. return instByID, err
  336. }
  337. instByID[s.ID] = inst.InstitutionID
  338. rep.InstitutionsCreated++
  339. }
  340. return instByID, nil
  341. }
  342. // writeCampuses 写 t_campus(幂等:按 campus_name + institution 复用)。
  343. func (r *Runner) writeCampuses(ctx context.Context, tx *gorm.DB, campuses []LegacySchoolCampus,
  344. instByID map[int32]int64, rep *MigrateReport) (map[int32]int64, error) {
  345. campusByID := make(map[int32]int64, len(campuses))
  346. for i := range campuses {
  347. c := &campuses[i]
  348. // 孤儿校区(school_id 不在被引用机构集):institution_id 置 0(机构级/平台级)
  349. instID, ok := instByID[c.SchoolID]
  350. if !ok {
  351. instID = 0
  352. }
  353. var existing []orgmodel.Campus
  354. tx.WithContext(ctx).Model(&orgmodel.Campus{}).
  355. Where("campus_name = ? AND institution_id = ?", c.Name, instID).Limit(1).Find(&existing)
  356. if len(existing) > 0 {
  357. campusByID[c.ID] = existing[0].CampusID
  358. continue
  359. }
  360. campus := &orgmodel.Campus{
  361. InstitutionID: instID,
  362. CampusName: c.Name,
  363. CampusCode: c.Code,
  364. Status: mapOrgStatus(c.Status),
  365. CreatedAt: r.opts.Now,
  366. UpdatedAt: r.opts.Now,
  367. }
  368. if err := tx.WithContext(ctx).Create(campus).Error; err != nil {
  369. return campusByID, err
  370. }
  371. campusByID[c.ID] = campus.CampusID
  372. rep.CampusesCreated++
  373. }
  374. return campusByID, nil
  375. }
  376. // writeUser 写 t_user(幂等:canonical_user_id 存在则跳过;username 冲突置 NULL)。
  377. func (r *Runner) writeUser(ctx context.Context, tx *gorm.DB, u *LegacyUser, canon string,
  378. d MatchDecision, rep *MigrateReport, usedUsername map[string]int32, existUsername map[string]int32) error {
  379. var n int64
  380. tx.WithContext(ctx).Model(&identitymodel.User{}).Where("canonical_user_id = ?", canon).Count(&n)
  381. if n > 0 {
  382. rep.UsersSkipped++
  383. return nil
  384. }
  385. // username:归一后若已占用(存量或本批首个外)→ 置 NULL + 记冲突
  386. var username *string
  387. if u.Username != "" {
  388. canonUsername := norm.Username(u.Username)
  389. if _, taken := usedUsername[canonUsername]; !taken {
  390. s := u.Username
  391. username = &s
  392. usedUsername[canonUsername] = u.ID
  393. } else {
  394. rep.UsernameConflicts = append(rep.UsernameConflicts, ConflictItem{
  395. LegacyID: u.ID,
  396. Detail: fmt.Sprintf("username=%q 归一后与已有用户重复,置 NULL", u.Username),
  397. })
  398. }
  399. }
  400. _ = existUsername // 存量 username 冲突已在 usedUsername 合并
  401. displayName := u.RealName
  402. user := &identitymodel.User{
  403. CanonicalUserID: canon,
  404. Username: username,
  405. DisplayName: displayName,
  406. AvatarURL: u.Avatar,
  407. PersonType: mapPersonType(u.Role),
  408. PasswordHash: &u.Password,
  409. PasswordSalt: &u.Salt,
  410. HashVersion: strPtr(identitymodel.HashVersionShaSalt), // I 库=教学侧,sha_salt(ADR-016 惰性升级)
  411. Status: mapUserStatus(u.Status),
  412. CreatedAt: r.opts.Now,
  413. UpdatedAt: r.opts.Now,
  414. }
  415. if u.RegTime != nil {
  416. user.CreatedAt = *u.RegTime
  417. }
  418. if err := tx.WithContext(ctx).Create(user).Error; err != nil {
  419. return err
  420. }
  421. rep.UsersCreated++
  422. if u.Status == 0 || u.Status == 3 {
  423. rep.UsersDisabled++
  424. }
  425. return nil
  426. }
  427. // writeIdentities 写 t_identity(phone/email/username/wechat;幂等 + 全局唯一键冲突处理)。
  428. func (r *Runner) writeIdentities(ctx context.Context, tx *gorm.DB, u *LegacyUser, canon string,
  429. d MatchDecision, rep *MigrateReport, usedPhone map[string]int32, existPhone map[string]int32) {
  430. // phone(强归一;撞号→本批首个占用,其余跳过+冲突已记)
  431. if u.PhoneNumber != nil && *u.PhoneNumber != "" {
  432. canonPhone, ok := norm.Phone(*u.PhoneNumber)
  433. if ok && !d.PhoneConflict {
  434. if _, taken := usedPhone[canonPhone]; !taken {
  435. if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypePhone, canonPhone, "", rep); err == nil {
  436. usedPhone[canonPhone] = u.ID
  437. rep.IdentitiesPhone++
  438. }
  439. }
  440. }
  441. }
  442. // email(轻归一)
  443. if u.Email != nil && *u.Email != "" {
  444. canonEmail, ok := norm.Email(*u.Email)
  445. if ok {
  446. if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeEmail, canonEmail, "", rep); err == nil {
  447. rep.IdentitiesEmail++
  448. }
  449. }
  450. }
  451. // username(轻归一;与 t_user.username 一致)
  452. if u.Username != "" {
  453. canonUsername := norm.Username(u.Username)
  454. if canonUsername != "" {
  455. if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeUsername, canonUsername, "", rep); err == nil {
  456. rep.IdentitiesUsername++
  457. }
  458. }
  459. }
  460. // wechat(openid + appid,经 t_weixin_user)
  461. if w := r.wechatOf(u.ID); w != nil {
  462. openid, ok := norm.WechatOpenID(w.OpenID)
  463. if ok {
  464. appid := ""
  465. if w.AppID != nil {
  466. appid = *w.AppID
  467. }
  468. if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeWechat, openid, appid, rep); err == nil {
  469. rep.IdentitiesWechat++
  470. }
  471. }
  472. }
  473. _ = existPhone
  474. }
  475. // wechatOf 取某 legacy user 的微信身份(runner 持有 planner 不方便,直接查 wx 缓存)。
  476. // 注:实现上在 Run 里通过闭包注入更合适;此处用字段缓存。
  477. func (r *Runner) wechatOf(userID int32) *LegacyWeixinUser {
  478. return r.wxCache[userID]
  479. }
  480. // upsertIdentity 幂等写入单条 identity(唯一键 type+identifier+appid 已存在则跳过)。
  481. func (r *Runner) upsertIdentity(ctx context.Context, tx *gorm.DB, canon, typ, identifier, appid string, rep *MigrateReport) error {
  482. var n int64
  483. tx.WithContext(ctx).Model(&identitymodel.Identity{}).
  484. Where("type = ? AND identifier = ? AND appid = ?", typ, identifier, appid).Count(&n)
  485. if n > 0 {
  486. rep.IdentityDuplicates++
  487. return nil
  488. }
  489. it := &identitymodel.Identity{
  490. CanonicalUserID: canon,
  491. Type: typ,
  492. Identifier: identifier,
  493. AppID: appid,
  494. Verified: false,
  495. CreatedAt: r.opts.Now,
  496. }
  497. return tx.WithContext(ctx).Create(it).Error
  498. }
  499. // writeWechatUnion 写 t_wechat_union(unionid 锚点;幂等:canonical/unionid 已存在则跳过)。
  500. func (r *Runner) writeWechatUnion(ctx context.Context, tx *gorm.DB, userID int32, canon string,
  501. planner *Planner, rep *MigrateReport) {
  502. w := r.wxCache[userID]
  503. if w == nil || w.UnionID == nil || *w.UnionID == "" {
  504. return
  505. }
  506. // 同 unionid 可能挂多个 canonical(跨 appid 同人):已存在任一 canonical 的该 unionid → 跳过
  507. var n int64
  508. tx.WithContext(ctx).Model(&identitymodel.WechatUnion{}).Where("unionid = ?", *w.UnionID).Count(&n)
  509. if n > 0 {
  510. return
  511. }
  512. wu := &identitymodel.WechatUnion{
  513. CanonicalUserID: canon,
  514. UnionID: *w.UnionID,
  515. CreatedAt: r.opts.Now,
  516. UpdatedAt: r.opts.Now,
  517. }
  518. if err := tx.WithContext(ctx).Create(wu).Error; err == nil {
  519. rep.WechatUnionCreated++
  520. } else {
  521. rep.Warnings = append(rep.Warnings, fmt.Sprintf("写 t_wechat_union unionid=%s 失败: %v", *w.UnionID, err))
  522. }
  523. }
  524. // writeLegacyMapping 写 t_legacy_mapping(canonical_user_id UNIQUE 幂等;重跑复用 ULID 时天然存在则跳过)。
  525. func (r *Runner) writeLegacyMapping(ctx context.Context, tx *gorm.DB, u *LegacyUser, canon string,
  526. d MatchDecision, rep *MigrateReport) error {
  527. var n int64
  528. tx.WithContext(ctx).Model(&identitymodel.LegacyMapping{}).Where("canonical_user_id = ?", canon).Count(&n)
  529. if n > 0 {
  530. return nil // 已存在(重跑幂等)
  531. }
  532. legacyID := u.ID
  533. role := mapRoleCode(u.Role)
  534. lm := &identitymodel.LegacyMapping{
  535. CanonicalUserID: canon,
  536. LegacyIUserID: &legacyID, // I 库=教学侧,legacy_i_user_id
  537. InstitutionID: 0, // 平台级/待 P1-05 细化(canonical 用户跨机构)
  538. CampusID: 0,
  539. Role: &role,
  540. MatchSource: strPtr(d.MatchSource),
  541. MatchConfidence: d.Confidence,
  542. AuditRef: strPtr(fmt.Sprintf("migrate/v1.2/%d", u.ID)),
  543. MergedAt: &r.opts.Now,
  544. CreatedAt: r.opts.Now,
  545. UpdatedAt: r.opts.Now,
  546. }
  547. if err := tx.WithContext(ctx).Create(lm).Error; err != nil {
  548. return err
  549. }
  550. rep.LegacyMappingsCreated++
  551. return nil
  552. }
  553. // writeMemberships 写 t_membership(幂等:唯一键 canonical+campus+role)。
  554. func (r *Runner) writeMemberships(ctx context.Context, tx *gorm.DB, binds []LegacyUserSchoolCampus,
  555. canonByID map[int32]string, instByID map[int32]int64, campusByID map[int32]int64,
  556. users []LegacyUser, rep *MigrateReport) {
  557. userByID := make(map[int32]*LegacyUser, len(users))
  558. for i := range users {
  559. userByID[users[i].ID] = &users[i]
  560. }
  561. for i := range binds {
  562. b := &binds[i]
  563. u, ok := userByID[b.UserID]
  564. if !ok {
  565. // 绑定用户不在 t_user(孤儿)→ 跳过 + 警告
  566. rep.Warnings = append(rep.Warnings, fmt.Sprintf("t_membership 跳过:legacy 绑定 userid=%d 不在 t_user", b.UserID))
  567. continue
  568. }
  569. canon, ok := canonByID[b.UserID]
  570. if !ok {
  571. continue
  572. }
  573. instID, ok1 := instByID[b.SchoolID]
  574. campusID, ok2 := campusByID[b.CampusID]
  575. if !ok1 || !ok2 {
  576. rep.Warnings = append(rep.Warnings, fmt.Sprintf("t_membership 跳过:institution/campus 映射缺失 schoolid=%d campusid=%d", b.SchoolID, b.CampusID))
  577. continue
  578. }
  579. role := mapRoleCode(u.Role)
  580. // 幂等:唯一键 (canonical, campus, role)
  581. var n int64
  582. tx.WithContext(ctx).Model(&orgmodel.Membership{}).
  583. Where("canonical_user_id = ? AND campus_id = ? AND role = ?", canon, campusID, role).Count(&n)
  584. if n > 0 {
  585. continue
  586. }
  587. m := &orgmodel.Membership{
  588. CanonicalUserID: canon,
  589. InstitutionID: instID,
  590. CampusID: campusID,
  591. Role: role,
  592. Status: orgmodel.MembershipStatusActive,
  593. JoinedAt: &r.opts.Now,
  594. CreatedAt: r.opts.Now,
  595. }
  596. if err := tx.WithContext(ctx).Create(m).Error; err == nil {
  597. rep.MembershipsCreated++
  598. } else {
  599. rep.Warnings = append(rep.Warnings, fmt.Sprintf("写 t_membership legacy_userid=%d 失败: %v", b.UserID, err))
  600. }
  601. }
  602. }
  603. // simulateCounts dry-run 计数预检(只读,不写):估算各表新建量。
  604. func (r *Runner) simulateCounts(ctx context.Context, rep *MigrateReport, schools []LegacySchool,
  605. campuses []LegacySchoolCampus, users []LegacyUser, canonByID map[int32]string, binds []LegacyUserSchoolCampus) {
  606. existingUsers := r.countTable(ctx, "t_user")
  607. existingInst := r.countTable(ctx, "t_institution")
  608. existingCampus := r.countTable(ctx, "t_campus")
  609. existingMapping := r.countTable(ctx, "t_legacy_mapping")
  610. // 幂等估算:新建 = 目标量 - 已存在量(取下限 0)
  611. if len(users) > int(existingUsers) {
  612. rep.UsersCreated = len(users) - int(existingUsers)
  613. }
  614. if len(schools) > int(existingInst) {
  615. rep.InstitutionsCreated = len(schools) - int(existingInst)
  616. }
  617. if len(campuses) > int(existingCampus) {
  618. rep.CampusesCreated = len(campuses) - int(existingCampus)
  619. }
  620. if len(users) > int(existingMapping) {
  621. rep.LegacyMappingsCreated = len(users) - int(existingMapping)
  622. }
  623. // 身份/成员/锚点按"全量首次写入"估算(重跑时这些由唯一键去重,量趋 0)
  624. for i := range users {
  625. u := &users[i]
  626. _ = canonByID[u.ID]
  627. if u.PhoneNumber != nil && *u.PhoneNumber != "" {
  628. if _, ok := norm.Phone(*u.PhoneNumber); ok {
  629. rep.IdentitiesPhone++
  630. }
  631. }
  632. if u.Email != nil && *u.Email != "" {
  633. if _, ok := norm.Email(*u.Email); ok {
  634. rep.IdentitiesEmail++
  635. }
  636. }
  637. if u.Username != "" {
  638. rep.IdentitiesUsername++
  639. }
  640. }
  641. for i := range users {
  642. if w := r.wxCache[users[i].ID]; w != nil {
  643. rep.IdentitiesWechat++
  644. if w.UnionID != nil && *w.UnionID != "" {
  645. rep.WechatUnionCreated++
  646. }
  647. }
  648. }
  649. rep.MembershipsCreated = len(binds)
  650. }
  651. // countTable 统计表行数(dry-run 幂等估算用)。
  652. func (r *Runner) countTable(ctx context.Context, table string) int64 {
  653. var n int64
  654. _ = r.target.WithContext(ctx).Table(table).Count(&n).Error
  655. return n
  656. }
  657. // ---- 映射函数 ----
  658. // mapUserStatus legacy status int → 目标态 string。
  659. // 1=active,0/2/3=disabled(P0-02 §4 拍板:0/3 都迁 disabled 保留可追溯)。
  660. func mapUserStatus(legacy int32) string {
  661. if legacy == 1 {
  662. return identitymodel.UserStatusActive
  663. }
  664. return identitymodel.UserStatusDisabled
  665. }
  666. // mapOrgStatus 机构/校区 legacy status(tinyint)→ 目标态 string。
  667. func mapOrgStatus(legacy int8) string {
  668. if legacy == 1 {
  669. return orgmodel.StatusActive
  670. }
  671. return orgmodel.StatusDisabled
  672. }
  673. // mapPersonType legacy role 数字码 → 目标态 person_type(P0-02 §3.1)。
  674. func mapPersonType(legacy int32) string {
  675. switch legacy {
  676. case 2:
  677. return identitymodel.PersonTypeStudent
  678. case 4:
  679. return identitymodel.PersonTypeTeacher
  680. case 3:
  681. return identitymodel.PersonTypeParent
  682. default:
  683. return identitymodel.PersonTypeStaff
  684. }
  685. }
  686. // mapRoleCode legacy role 数字码 → 目标态 7 角色字符串码(P0-02 §2.5 拍板映射)。
  687. func mapRoleCode(legacy int32) string {
  688. switch legacy {
  689. case 1:
  690. return orgmodel.RoleSuperAdmin
  691. case 2:
  692. return orgmodel.RoleStudent
  693. case 3:
  694. return orgmodel.RoleParent
  695. case 4:
  696. return orgmodel.RoleTeacher
  697. case 5, 6:
  698. return orgmodel.RoleAcademic // admin/redactor → 教务
  699. case 8:
  700. return orgmodel.RoleFinance
  701. case 7, 9:
  702. return orgmodel.RoleMarket // prototype/marketer → 市场
  703. default:
  704. return orgmodel.RoleAcademic // 未知角色兜底为教务(运营角色)
  705. }
  706. }
  707. // strPtr 字符串指针辅助。
  708. func strPtr(s string) *string { return &s }
  709. // gormMySQLTarget 目标库 GORM 驱动(mysql.Open 返回 Dialector,与 infra/db 同款)。
  710. func gormMySQLTarget(dsn string) gorm.Dialector {
  711. return mysql.Open(dsn)
  712. }