Browse Source

fix(user-migration): revalidate drift and verify synced quota

master
fengsilin 1 week ago
parent
commit
3a78e9c93a
5 changed files with 476 additions and 3 deletions
  1. +1
    -0
      controller/internal_user_migration.go
  2. +68
    -0
      service/user_migration/executor.go
  3. +401
    -0
      service/user_migration/service_test.go
  4. +1
    -0
      service/user_migration/types.go
  5. +5
    -3
      service/user_migration/verifier.go

+ 1
- 0
controller/internal_user_migration.go View File

@@ -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)


+ 68
- 0
service/user_migration/executor.go View File

@@ -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,


+ 401
- 0
service/user_migration/service_test.go View File

@@ -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)
}

+ 1
- 0
service/user_migration/types.go View File

@@ -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 {


+ 5
- 3
service/user_migration/verifier.go View File

@@ -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
}

Loading…
Cancel
Save