From 3a78e9c93af6926e1aab8865839c25d9cc029132 Mon Sep 17 00:00:00 2001 From: fengsilin Date: Wed, 3 Jun 2026 13:37:39 +0800 Subject: [PATCH] fix(user-migration): revalidate drift and verify synced quota --- controller/internal_user_migration.go | 1 + service/user_migration/executor.go | 68 +++++ service/user_migration/service_test.go | 401 +++++++++++++++++++++++++ service/user_migration/types.go | 1 + service/user_migration/verifier.go | 8 +- 5 files changed, 476 insertions(+), 3 deletions(-) diff --git a/controller/internal_user_migration.go b/controller/internal_user_migration.go index f86c4ea..d471482 100644 --- a/controller/internal_user_migration.go +++ b/controller/internal_user_migration.go @@ -109,6 +109,7 @@ func buildRemoteUserSnapshot(user *model.User, bindings []*model.UserOAuthBindin LinuxDOId: user.LinuxDOId, Source: user.Source, RemoteUserId: user.RemoteUserId, + SyncedQuota: user.SyncedQuota, } for _, binding := range bindings { provider, err := model.GetCustomOAuthProviderById(binding.ProviderId) diff --git a/service/user_migration/executor.go b/service/user_migration/executor.go index 51dcb80..91b3f98 100644 --- a/service/user_migration/executor.go +++ b/service/user_migration/executor.go @@ -53,6 +53,12 @@ func (s *Service) executeCreateNew(item *model.UserMigrationItem) error { if err != nil { return fmt.Errorf("get remote user %d: %w", item.SourceUserId, err) } + if err := s.revalidateCreateNew(item, snapshot); err != nil { + return err + } + if err := markItemMigrating(item.Id); err != nil { + return err + } affCode := snapshot.AffCode if affCode != "" { @@ -112,6 +118,12 @@ func (s *Service) executeMergeExisting(item *model.UserMigrationItem) error { if err != nil { return fmt.Errorf("get remote user %d: %w", item.SourceUserId, err) } + if err := s.revalidateMergeExisting(item, snapshot); err != nil { + return err + } + if err := markItemMigrating(item.Id); err != nil { + return err + } finalQuota := 0 err = model.DB.Transaction(func(tx *gorm.DB) error { @@ -185,6 +197,62 @@ func (s *Service) ensureRemoteConverted(item *model.UserMigrationItem, syncedQuo return markItemMigrated(item.Id) } +func (s *Service) revalidateCreateNew(item *model.UserMigrationItem, snapshot *RemoteUserSnapshot) error { + if snapshot.Source == common.UserSourceSynced { + return fmt.Errorf("execution-time identity drift detected, remote user %d is already synced, rescan required", item.SourceUserId) + } + + result, err := s.analyzeExecutionSnapshot(snapshot) + if err != nil { + return err + } + if result.Status != model.UserMigrationItemStatusReady || + result.Strategy != model.UserMigrationStrategyCreateNew || + result.TargetCNUserID != 0 { + return fmt.Errorf("execution-time identity drift detected, rescan required") + } + return nil +} + +func (s *Service) revalidateMergeExisting(item *model.UserMigrationItem, snapshot *RemoteUserSnapshot) error { + if snapshot.Source == common.UserSourceSynced { + if snapshot.RemoteUserId == item.TargetUserId { + return nil + } + return fmt.Errorf("execution-time identity drift detected, remote user %d is already synced to CN user %d", item.SourceUserId, snapshot.RemoteUserId) + } + + result, err := s.analyzeExecutionSnapshot(snapshot) + if err != nil { + return err + } + if result.Status != model.UserMigrationItemStatusReady || + result.Strategy != model.UserMigrationStrategyMergeExisting || + result.TargetCNUserID != item.TargetUserId { + return fmt.Errorf("execution-time merge target drift detected, rescan required") + } + return nil +} + +func (s *Service) analyzeExecutionSnapshot(snapshot *RemoteUserSnapshot) (*conflictResult, error) { + providerSlugMap, err := buildProviderSlugMap() + if err != nil { + return nil, err + } + result, err := analyzeConflicts(snapshot, providerSlugMap, s.client) + if err != nil { + return nil, fmt.Errorf("revalidate conflicts for user %d: %w", snapshot.Id, err) + } + return result, nil +} + +func markItemMigrating(itemID int) error { + return model.DB.Model(&model.UserMigrationItem{}).Where("id = ?", itemID).Updates(map[string]any{ + "status": model.UserMigrationItemStatusMigrating, + "error_message": "", + }).Error +} + func markItemMigrated(itemID int) error { return model.DB.Model(&model.UserMigrationItem{}).Where("id = ?", itemID).Updates(map[string]any{ "status": model.UserMigrationItemStatusMigrated, diff --git a/service/user_migration/service_test.go b/service/user_migration/service_test.go index fffe51c..e6ce378 100644 --- a/service/user_migration/service_test.go +++ b/service/user_migration/service_test.go @@ -51,10 +51,12 @@ func (f *fakeClient) ConvertToSynced(userId int, req *ConvertRemoteUserRequest) if snapshot, ok := f.getUserByID[userId]; ok && snapshot != nil { snapshot.Source = common.UserSourceSynced snapshot.RemoteUserId = req.RemoteUserId + snapshot.SyncedQuota = req.SyncedQuota } if f.getUserResp != nil && f.getUserResp.Id == userId { f.getUserResp.Source = common.UserSourceSynced f.getUserResp.RemoteUserId = req.RemoteUserId + f.getUserResp.SyncedQuota = req.SyncedQuota } } return nil @@ -284,6 +286,227 @@ func TestScanBatch_Rescan_PreservesHumanStrategy(t *testing.T) { require.Equal(t, model.UserMigrationStrategySkip, items[0].ResolutionStrategy) } +func TestMigrationFlow_GrayScenarios_EndToEnd(t *testing.T) { + db := setupMigrationServiceDB(t) + orig := model.DB + model.DB = db + defer func() { + model.DB = orig + }() + + require.NoError(t, db.Create(&model.User{ + Id: 1, + Username: "cn-merge", + Password: "hash", + Email: "merge@example.com", + AffCode: "CNM1", + Quota: 500, + }).Error) + require.NoError(t, db.Create(&model.User{ + Id: 2, + Username: "taken-name", + Password: "hash", + Email: "taken@example.com", + AffCode: "CNT2", + Quota: 50, + }).Error) + require.NoError(t, db.Create(&model.User{ + Id: 3, + Username: "cn-hard", + Password: "hash", + Email: "synced@example.com", + AffCode: "CNH3", + Quota: 300, + }).Error) + require.NoError(t, db.Create(&model.User{ + Id: 4, + Username: "cn-soft", + Password: "hash", + Email: "soft-owner@example.com", + AffCode: "SOFT1", + Quota: 80, + }).Error) + + batch := &model.UserMigrationBatch{ + Name: "gray-scenarios", + SourceRegion: "ov", + TargetRegion: "cn", + OperatorId: 1, + } + require.NoError(t, model.CreateUserMigrationBatch(batch)) + + remoteUsers := []RemoteUserSnapshot{ + { + Id: 10000001, + Username: "ov-create", + Email: "create@example.com", + Quota: 100, + AffCode: "OVC1", + }, + { + Id: 10000002, + Username: "ov-merge", + Email: "merge@example.com", + Quota: 200, + AffCode: "OVM2", + }, + { + Id: 10000003, + Username: "ov-hard", + Email: "synced@example.com", + Quota: 300, + AffCode: "OVH3", + }, + { + Id: 10000004, + Username: "taken-name", + Email: "username-conflict@example.com", + Quota: 120, + AffCode: "OVU4", + }, + { + Id: 10000005, + Username: "ov-soft", + Email: "soft@example.com", + Quota: 150, + AffCode: "SOFT1", + }, + } + + getUserByID := make(map[int]*RemoteUserSnapshot, len(remoteUsers)) + for i := range remoteUsers { + snapshot := remoteUsers[i] + getUserByID[snapshot.Id] = &snapshot + } + + client := &fakeClient{ + listPages: map[int][]RemoteUserSnapshot{ + 0: remoteUsers, + }, + getUserByID: getUserByID, + hasCopyByID: map[int]bool{ + 1: false, + 3: true, + }, + } + svc := &Service{client: client} + + require.NoError(t, svc.ScanBatch(batch.Id)) + + items, err := model.ListUserMigrationItemsByBatch(batch.Id) + require.NoError(t, err) + require.Len(t, items, 5) + + bySourceUserID := make(map[int]*model.UserMigrationItem, len(items)) + for _, item := range items { + bySourceUserID[item.SourceUserId] = item + } + + require.Equal(t, model.UserMigrationStrategyCreateNew, bySourceUserID[10000001].ResolutionStrategy) + require.Equal(t, model.UserMigrationItemStatusReady, bySourceUserID[10000001].Status) + + require.Equal(t, model.UserMigrationStrategyMergeExisting, bySourceUserID[10000002].ResolutionStrategy) + require.Equal(t, 1, bySourceUserID[10000002].TargetUserId) + require.Equal(t, model.UserMigrationItemStatusReady, bySourceUserID[10000002].Status) + + require.Equal(t, model.UserMigrationItemStatusConflict, bySourceUserID[10000003].Status) + require.Contains(t, bySourceUserID[10000003].ConflictFlags, "cn_already_synced_to_ov") + + require.Equal(t, model.UserMigrationItemStatusConflict, bySourceUserID[10000004].Status) + require.Contains(t, bySourceUserID[10000004].ConflictFlags, "username") + + require.Equal(t, model.UserMigrationStrategyCreateNew, bySourceUserID[10000005].ResolutionStrategy) + require.Equal(t, model.UserMigrationItemStatusReady, bySourceUserID[10000005].Status) + require.Contains(t, bySourceUserID[10000005].ConflictFlags, "aff_code") + + require.NoError(t, db.Model(&model.UserMigrationItem{}). + Where("id IN ?", []int{bySourceUserID[10000003].Id, bySourceUserID[10000004].Id}). + Updates(map[string]any{ + "resolution_strategy": model.UserMigrationStrategySkip, + "status": model.UserMigrationItemStatusReady, + "error_message": "", + }).Error) + require.NoError(t, model.RefreshUserMigrationBatchStats(batch.Id)) + + var readyBatch model.UserMigrationBatch + require.NoError(t, db.First(&readyBatch, batch.Id).Error) + require.Equal(t, model.UserMigrationBatchStatusReady, readyBatch.Status) + + locked, err := model.TryLockBatchForExecution(batch.Id) + require.NoError(t, err) + require.True(t, locked) + require.NoError(t, svc.ExecuteBatch(batch.Id)) + + items, err = model.ListUserMigrationItemsByBatch(batch.Id) + require.NoError(t, err) + bySourceUserID = make(map[int]*model.UserMigrationItem, len(items)) + for _, item := range items { + bySourceUserID[item.SourceUserId] = item + } + + require.Equal(t, model.UserMigrationItemStatusMigrated, bySourceUserID[10000001].Status) + require.Equal(t, model.UserMigrationItemStatusMigrated, bySourceUserID[10000002].Status) + require.Equal(t, model.UserMigrationItemStatusSkipped, bySourceUserID[10000003].Status) + require.Equal(t, model.UserMigrationItemStatusSkipped, bySourceUserID[10000004].Status) + require.Equal(t, model.UserMigrationItemStatusMigrated, bySourceUserID[10000005].Status) + + var createdUser model.User + require.NoError(t, db.Where("username = ?", "ov-create").First(&createdUser).Error) + require.Equal(t, 100, createdUser.Quota) + + var mergedUser model.User + require.NoError(t, db.First(&mergedUser, 1).Error) + require.Equal(t, 700, mergedUser.Quota) + + var softUser model.User + require.NoError(t, db.Where("username = ?", "ov-soft").First(&softUser).Error) + require.Equal(t, 150, softUser.Quota) + require.NotEqual(t, "SOFT1", softUser.AffCode) + + require.Equal(t, common.UserSourceSynced, client.getUserByID[10000001].Source) + require.Equal(t, createdUser.Id, client.getUserByID[10000001].RemoteUserId) + require.Equal(t, 100, client.getUserByID[10000001].SyncedQuota) + + require.Equal(t, common.UserSourceSynced, client.getUserByID[10000002].Source) + require.Equal(t, 1, client.getUserByID[10000002].RemoteUserId) + require.Equal(t, 700, client.getUserByID[10000002].SyncedQuota) + + require.NotEqual(t, common.UserSourceSynced, client.getUserByID[10000003].Source) + require.NotEqual(t, common.UserSourceSynced, client.getUserByID[10000004].Source) + + require.Equal(t, common.UserSourceSynced, client.getUserByID[10000005].Source) + require.Equal(t, softUser.Id, client.getUserByID[10000005].RemoteUserId) + require.Equal(t, 150, client.getUserByID[10000005].SyncedQuota) + + for _, sourceUserID := range []int{10000001, 10000002, 10000005} { + result, verifyErr := svc.VerifyItem(bySourceUserID[sourceUserID]) + require.NoError(t, verifyErr) + require.True(t, result.TargetUserExists) + require.True(t, result.RemoteConverted) + require.True(t, result.QuotaMatched) + } + + require.NoError(t, db.Model(&model.UserMigrationBatch{}).Where("id = ?", batch.Id). + Update("status", model.UserMigrationBatchStatusReady).Error) + require.NoError(t, db.Model(&model.UserMigrationItem{}). + Where("id IN ?", []int{ + bySourceUserID[10000001].Id, + bySourceUserID[10000002].Id, + bySourceUserID[10000005].Id, + }). + Update("status", model.UserMigrationItemStatusReady).Error) + + locked, err = model.TryLockBatchForExecution(batch.Id) + require.NoError(t, err) + require.True(t, locked) + require.NoError(t, svc.ExecuteBatch(batch.Id)) + + require.Len(t, client.convertCalls, 3) + + require.NoError(t, db.First(&mergedUser, 1).Error) + require.Equal(t, 700, mergedUser.Quota) +} + func TestExecuteBatch_CreateNew_CreatesUserAndConvertsRemote(t *testing.T) { db := setupMigrationServiceDB(t) orig := model.DB @@ -404,6 +627,74 @@ func TestExecuteBatch_CreateNew_IdempotentOnRetry(t *testing.T) { require.Len(t, client.convertCalls, 1) } +func TestExecuteBatch_CreateNew_ExecutionTimeDriftFails(t *testing.T) { + db := setupMigrationServiceDB(t) + orig := model.DB + model.DB = db + defer func() { + model.DB = orig + }() + + require.NoError(t, db.Create(&model.User{ + Id: 7, + Username: "cn-alice", + Password: "hash", + Email: "alice@example.com", + AffCode: "C007", + }).Error) + + batch := &model.UserMigrationBatch{ + Name: "wave-create-drift", + SourceRegion: "ov", + TargetRegion: "cn", + OperatorId: 1, + Status: model.UserMigrationBatchStatusReady, + } + require.NoError(t, model.CreateUserMigrationBatch(batch)) + + item := &model.UserMigrationItem{ + BatchId: batch.Id, + SourceUserId: 10000020, + SourceUsername: "ov-drift", + SourceEmail: "old@example.com", + SourceQuota: 120, + ResolutionStrategy: model.UserMigrationStrategyCreateNew, + Status: model.UserMigrationItemStatusReady, + MigrationToken: "batch-1-user-10000020", + } + require.NoError(t, model.UpsertUserMigrationItem(item)) + + client := &fakeClient{ + getUserResp: &RemoteUserSnapshot{ + Id: 10000020, + Username: "ov-drift", + Password: "hash", + Email: "alice@example.com", + Quota: 120, + AffCode: "OV20", + }, + hasCopyByID: map[int]bool{7: false}, + } + svc := &Service{client: client} + + locked, err := model.TryLockBatchForExecution(batch.Id) + require.NoError(t, err) + require.True(t, locked) + require.NoError(t, svc.ExecuteBatch(batch.Id)) + + require.Len(t, client.convertCalls, 0) + + var count int64 + require.NoError(t, db.Model(&model.User{}).Where("username = ?", "ov-drift").Count(&count).Error) + require.EqualValues(t, 0, count) + + items, err := model.ListUserMigrationItemsByBatch(batch.Id) + require.NoError(t, err) + require.Len(t, items, 1) + require.Equal(t, model.UserMigrationItemStatusFailed, items[0].Status) + require.Contains(t, items[0].ErrorMessage, "identity drift") +} + func TestExecuteBatch_MergeExisting_IdempotentQuota(t *testing.T) { db := setupMigrationServiceDB(t) orig := model.DB @@ -472,6 +763,75 @@ func TestExecuteBatch_MergeExisting_IdempotentQuota(t *testing.T) { require.Len(t, client.convertCalls, 1) } +func TestExecuteBatch_MergeExisting_ExecutionTimeDriftFails(t *testing.T) { + db := setupMigrationServiceDB(t) + orig := model.DB + model.DB = db + defer func() { + model.DB = orig + }() + + require.NoError(t, db.Create(&model.User{ + Id: 1, + Username: "cn-bob", + Password: "hash", + Email: "bob@example.com", + AffCode: "CNB1", + Quota: 500, + }).Error) + + batch := &model.UserMigrationBatch{ + Name: "wave-merge-drift", + SourceRegion: "ov", + TargetRegion: "cn", + OperatorId: 1, + Status: model.UserMigrationBatchStatusReady, + } + require.NoError(t, model.CreateUserMigrationBatch(batch)) + + item := &model.UserMigrationItem{ + BatchId: batch.Id, + SourceUserId: 10000021, + TargetUserId: 1, + SourceUsername: "ov-bob", + SourceEmail: "bob@example.com", + SourceQuota: 200, + ResolutionStrategy: model.UserMigrationStrategyMergeExisting, + Status: model.UserMigrationItemStatusReady, + MigrationToken: "batch-1-user-10000021", + } + require.NoError(t, model.UpsertUserMigrationItem(item)) + + client := &fakeClient{ + getUserResp: &RemoteUserSnapshot{ + Id: 10000021, + Username: "ov-bob", + Password: "hash", + Email: "changed@example.com", + Quota: 200, + AffCode: "OV21", + }, + } + svc := &Service{client: client} + + locked, err := model.TryLockBatchForExecution(batch.Id) + require.NoError(t, err) + require.True(t, locked) + require.NoError(t, svc.ExecuteBatch(batch.Id)) + + require.Len(t, client.convertCalls, 0) + + var cnUser model.User + require.NoError(t, db.First(&cnUser, 1).Error) + require.Equal(t, 500, cnUser.Quota) + + items, err := model.ListUserMigrationItemsByBatch(batch.Id) + require.NoError(t, err) + require.Len(t, items, 1) + require.Equal(t, model.UserMigrationItemStatusFailed, items[0].Status) + require.Contains(t, items[0].ErrorMessage, "merge target drift") +} + func TestExecuteBatch_SkipItem_DoesNotConvert(t *testing.T) { db := setupMigrationServiceDB(t) orig := model.DB @@ -542,6 +902,7 @@ func TestVerifyItem_CorrectlyChecksRemoteConverted(t *testing.T) { Id: 10000001, Source: common.UserSourceSynced, RemoteUserId: 789, + SyncedQuota: 300, }, } svc := &Service{client: client} @@ -551,3 +912,43 @@ func TestVerifyItem_CorrectlyChecksRemoteConverted(t *testing.T) { require.True(t, result.RemoteConverted) require.True(t, result.QuotaMatched) } + +func TestVerifyItem_QuotaMismatchDetected(t *testing.T) { + db := setupMigrationServiceDB(t) + orig := model.DB + model.DB = db + defer func() { + model.DB = orig + }() + + require.NoError(t, db.Create(&model.User{ + Id: 790, + Username: "cn-mismatch", + Password: "hash", + AffCode: "CM2", + Quota: 300, + }).Error) + + item := &model.UserMigrationItem{ + BatchId: 1, + SourceUserId: 10000002, + TargetUserId: 790, + SourceQuota: 300, + Status: model.UserMigrationItemStatusMigrated, + } + + client := &fakeClient{ + getUserResp: &RemoteUserSnapshot{ + Id: 10000002, + Source: common.UserSourceSynced, + RemoteUserId: 790, + SyncedQuota: 250, + }, + } + svc := &Service{client: client} + result, err := svc.VerifyItem(item) + require.NoError(t, err) + require.True(t, result.TargetUserExists) + require.True(t, result.RemoteConverted) + require.False(t, result.QuotaMatched) +} diff --git a/service/user_migration/types.go b/service/user_migration/types.go index 1605a22..050aab1 100644 --- a/service/user_migration/types.go +++ b/service/user_migration/types.go @@ -27,6 +27,7 @@ type RemoteUserSnapshot struct { OAuthBindings []RemoteOAuthBinding `json:"oauth_bindings"` Source string `json:"source"` RemoteUserId int `json:"remote_user_id"` + SyncedQuota int `json:"synced_quota"` } type ListRemoteUsersResponse struct { diff --git a/service/user_migration/verifier.go b/service/user_migration/verifier.go index 229d70b..e87fa6a 100644 --- a/service/user_migration/verifier.go +++ b/service/user_migration/verifier.go @@ -22,10 +22,12 @@ func (s *Service) VerifyItem(item *model.UserMigrationItem) (*VerifyResult, erro return nil, err } + remoteConverted := remoteUser.Source == common.UserSourceSynced && + remoteUser.RemoteUserId == item.TargetUserId + return &VerifyResult{ TargetUserExists: targetUser != nil, - RemoteConverted: remoteUser.Source == common.UserSourceSynced && - remoteUser.RemoteUserId == item.TargetUserId, - QuotaMatched: targetUser != nil && targetUser.Quota >= item.SourceQuota, + RemoteConverted: remoteConverted, + QuotaMatched: targetUser != nil && remoteConverted && remoteUser.SyncedQuota == targetUser.Quota, }, nil }