|
|
@@ -237,6 +237,9 @@ func (r *Runner) Run(ctx context.Context) (*MigrateReport, error) {
|
|
|
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)
|
|
|
@@ -247,23 +250,30 @@ func (r *Runner) Run(ctx context.Context) (*MigrateReport, error) {
|
|
|
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, existUsername); err != nil {
|
|
|
+ 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, existPhone)
|
|
|
- r.writeWechatUnion(ctx, tx, u.ID, canon, planner, rep)
|
|
|
- if err := r.writeLegacyMapping(ctx, tx, u, canon, d, teachingByUserID, rep); err != nil {
|
|
|
+ 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, rep)
|
|
|
+ r.writeMemberships(ctx, tx, binds, canonByID, instByID, campusByID, users, existMemberKeys, rep)
|
|
|
return nil
|
|
|
})
|
|
|
if err != nil {
|
|
|
@@ -346,6 +356,85 @@ func (r *Runner) existingIdentities(ctx context.Context, tx *gorm.DB, typ, appid
|
|
|
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 查现有复用)。
|
|
|
@@ -410,12 +499,12 @@ func (r *Runner) writeCampuses(ctx context.Context, tx *gorm.DB, campuses []Lega
|
|
|
return campusByID, nil
|
|
|
}
|
|
|
|
|
|
-// writeUser 写 t_user(幂等:canonical_user_id 存在则跳过;username 冲突置 NULL)。
|
|
|
+// 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, existUsername map[string]int32) error {
|
|
|
- var n int64
|
|
|
- tx.WithContext(ctx).Model(&identitymodel.User{}).Where("canonical_user_id = ?", canon).Count(&n)
|
|
|
- if n > 0 {
|
|
|
+ d MatchDecision, rep *MigrateReport, usedUsername map[string]int32, existUserCanons map[string]struct{}) error {
|
|
|
+ if _, ok := existUserCanons[canon]; ok {
|
|
|
rep.UsersSkipped++
|
|
|
return nil
|
|
|
}
|
|
|
@@ -434,7 +523,6 @@ func (r *Runner) writeUser(ctx context.Context, tx *gorm.DB, u *LegacyUser, cano
|
|
|
})
|
|
|
}
|
|
|
}
|
|
|
- _ = existUsername // 存量 username 冲突已在 usedUsername 合并
|
|
|
|
|
|
displayName := u.RealName
|
|
|
user := &identitymodel.User{
|
|
|
@@ -456,6 +544,7 @@ func (r *Runner) writeUser(ctx context.Context, tx *gorm.DB, u *LegacyUser, cano
|
|
|
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++
|
|
|
@@ -464,14 +553,16 @@ func (r *Runner) writeUser(ctx context.Context, tx *gorm.DB, u *LegacyUser, cano
|
|
|
}
|
|
|
|
|
|
// 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, existPhone map[string]int32) {
|
|
|
+ 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, "", rep); err == nil {
|
|
|
+ if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypePhone, canonPhone, "", existIdentityKeys, rep); err == nil {
|
|
|
usedPhone[canonPhone] = u.ID
|
|
|
rep.IdentitiesPhone++
|
|
|
}
|
|
|
@@ -482,7 +573,7 @@ func (r *Runner) writeIdentities(ctx context.Context, tx *gorm.DB, u *LegacyUser
|
|
|
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 {
|
|
|
+ if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeEmail, canonEmail, "", existIdentityKeys, rep); err == nil {
|
|
|
rep.IdentitiesEmail++
|
|
|
}
|
|
|
}
|
|
|
@@ -491,7 +582,7 @@ func (r *Runner) writeIdentities(ctx context.Context, tx *gorm.DB, u *LegacyUser
|
|
|
if u.Username != "" {
|
|
|
canonUsername := norm.Username(u.Username)
|
|
|
if canonUsername != "" {
|
|
|
- if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeUsername, canonUsername, "", rep); err == nil {
|
|
|
+ if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeUsername, canonUsername, "", existIdentityKeys, rep); err == nil {
|
|
|
rep.IdentitiesUsername++
|
|
|
}
|
|
|
}
|
|
|
@@ -504,12 +595,11 @@ func (r *Runner) writeIdentities(ctx context.Context, tx *gorm.DB, u *LegacyUser
|
|
|
if w.AppID != nil {
|
|
|
appid = *w.AppID
|
|
|
}
|
|
|
- if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeWechat, openid, appid, rep); err == nil {
|
|
|
+ if err := r.upsertIdentity(ctx, tx, canon, identitymodel.IdentityTypeWechat, openid, appid, existIdentityKeys, rep); err == nil {
|
|
|
rep.IdentitiesWechat++
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
- _ = existPhone
|
|
|
}
|
|
|
|
|
|
// wechatOf 取某 legacy user 的微信身份(runner 持有 planner 不方便,直接查 wx 缓存)。
|
|
|
@@ -519,11 +609,12 @@ func (r *Runner) wechatOf(userID int32) *LegacyWeixinUser {
|
|
|
}
|
|
|
|
|
|
// 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 {
|
|
|
+//
|
|
|
+// 性能:幂等判断用 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
|
|
|
}
|
|
|
@@ -535,20 +626,24 @@ func (r *Runner) upsertIdentity(ctx context.Context, tx *gorm.DB, canon, typ, id
|
|
|
Verified: false,
|
|
|
CreatedAt: r.opts.Now,
|
|
|
}
|
|
|
- return tx.WithContext(ctx).Create(it).Error
|
|
|
+ if err := tx.WithContext(ctx).Create(it).Error; err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ existIdentityKeys[key] = struct{}{} // 标记已占用
|
|
|
+ return nil
|
|
|
}
|
|
|
|
|
|
-// writeWechatUnion 写 t_wechat_union(unionid 锚点;幂等:canonical/unionid 已存在则跳过)。
|
|
|
+// writeWechatUnion 写 t_wechat_union(unionid 锚点;幂等:unionid 已存在则跳过)。
|
|
|
+//
|
|
|
+// 性能:幂等判断用 existUnionids 内存 set(事务内一次性批量装载),不再逐行 COUNT。
|
|
|
func (r *Runner) writeWechatUnion(ctx context.Context, tx *gorm.DB, userID int32, canon string,
|
|
|
- planner *Planner, rep *MigrateReport) {
|
|
|
+ existUnionids map[string]struct{}, 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 {
|
|
|
+ // 同 unionid 可能挂多个 canonical(跨 appid 同人):已存在该 unionid → 跳过
|
|
|
+ if _, ok := existUnionids[*w.UnionID]; ok {
|
|
|
return
|
|
|
}
|
|
|
wu := &identitymodel.WechatUnion{
|
|
|
@@ -558,6 +653,7 @@ func (r *Runner) writeWechatUnion(ctx context.Context, tx *gorm.DB, userID int32
|
|
|
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))
|
|
|
@@ -571,11 +667,9 @@ func (r *Runner) writeWechatUnion(ctx context.Context, tx *gorm.DB, userID int32
|
|
|
// - 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, rep *MigrateReport) error {
|
|
|
- var n int64
|
|
|
- tx.WithContext(ctx).Model(&identitymodel.LegacyMapping{}).Where("canonical_user_id = ?", canon).Count(&n)
|
|
|
- if n > 0 {
|
|
|
- return nil // 已存在(重跑幂等)
|
|
|
+ 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)
|
|
|
@@ -599,14 +693,17 @@ func (r *Runner) writeLegacyMapping(ctx context.Context, tx *gorm.DB, u *LegacyU
|
|
|
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, rep *MigrateReport) {
|
|
|
+ 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]
|
|
|
@@ -631,10 +728,7 @@ func (r *Runner) writeMemberships(ctx context.Context, tx *gorm.DB, binds []Lega
|
|
|
}
|
|
|
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 {
|
|
|
+ if _, ok := existMemberKeys[memberKey(canon, campusID, role)]; ok {
|
|
|
continue
|
|
|
}
|
|
|
m := &orgmodel.Membership{
|
|
|
@@ -647,6 +741,7 @@ func (r *Runner) writeMemberships(ctx context.Context, tx *gorm.DB, binds []Lega
|
|
|
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))
|