package impl

import (
	"context"
	"testing"

	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	grpcli_mock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	regionalapi_mock "github.com/tidbcloud/dataflow-service/pkg/regional-api/mock"
	replay_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	mgmt_commonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

func prepareReplaySourceForTest(
	ctx context.Context,
	operator *replayOperatorImpl,
	replayEnt *entity.Replay,
	sourceCluster *mgmt.Cluster,
) bizerr.BizError {
	provider, bizErr := newReplayClusterProvider(operator, sourceCluster, replayEnt)
	if bizErr != nil || provider == nil {
		return bizErr
	}
	sourceAuditLogAddr, bizErr := provider.PrepareSourceAuditLog(ctx)
	if replayEnt != nil {
		replayEnt.SourceAuditCredential = sourceAuditLogAddr
	}
	return bizErr
}

func TestNormalizeAliyunAuditLogURL(t *testing.T) {
	t.Run("convert oss url to s3 compatible oss url", func(t *testing.T) {
		cred := &entity.AliyunAuditLog{
			BucketURL:       "oss://db-auditlog-fang-test/10480461862511845599/tidb-%2A",
			AccessKeyID:     "test-ak",
			SecretAccessKey: "test-sk",
		}

		err := normalizeAliyunAuditLogURL(cred, "ap-southeast-1")
		require.NoError(t, err)
		require.Equal(
			t,
			"s3://db-auditlog-fang-test/10480461862511845599/tidb-?access-key=test-ak&endpoint=http%3A%2F%2Foss-ap-southeast-1.aliyuncs.com&secret-access-key=test-sk",
			cred.BucketURL,
		)
	})

	t.Run("escape oss query values safely", func(t *testing.T) {
		cred := &entity.AliyunAuditLog{
			BucketURL:       "oss://db-auditlog-fang-test/10480461862511845599/tidb-%2A",
			AccessKeyID:     "test+ak&1",
			SecretAccessKey: "test?sk=2&x",
		}

		err := normalizeAliyunAuditLogURL(cred, "ap-southeast-1")
		require.NoError(t, err)
		require.Equal(
			t,
			"s3://db-auditlog-fang-test/10480461862511845599/tidb-?access-key=test%2Bak%261&endpoint=http%3A%2F%2Foss-ap-southeast-1.aliyuncs.com&secret-access-key=test%3Fsk%3D2%26x",
			cred.BucketURL,
		)
	})

	t.Run("pass through arbitrary url when keys are omitted", func(t *testing.T) {
		cred := &entity.AliyunAuditLog{
			BucketURL: "https://audit.example.com/db-auditlog-fang-test/10480461862511845599/tidb-%2A",
		}

		err := normalizeAliyunAuditLogURL(cred, "ap-southeast-1")
		require.NoError(t, err)
		require.Equal(
			t,
			"https://audit.example.com/db-auditlog-fang-test/10480461862511845599/tidb-%2A",
			cred.BucketURL,
		)
	})

	t.Run("reject partial key pair", func(t *testing.T) {
		cred := &entity.AliyunAuditLog{
			BucketURL:   "oss://db-auditlog-fang-test/10480461862511845599/tidb-%2A",
			AccessKeyID: "test-ak",
		}

		err := normalizeAliyunAuditLogURL(cred, "ap-southeast-1")
		require.Error(t, err)
		require.Contains(t, err.Error(), "access_key_id and secret_access_key must be provided together")
	})

	t.Run("pass through s3 url when keys are omitted", func(t *testing.T) {
		cred := &entity.AliyunAuditLog{
			BucketURL: "s3://db-auditlog-fang-test/10480461862511845599/tidb-*?endpoint=http://oss-ap-southeast-1.aliyuncs.com&access-key=test-ak&secret-access-key=test-sk",
		}

		err := normalizeAliyunAuditLogURL(cred, "ap-southeast-1")
		require.NoError(t, err)
		require.Equal(
			t,
			"s3://db-auditlog-fang-test/10480461862511845599/tidb-*?endpoint=http://oss-ap-southeast-1.aliyuncs.com&access-key=test-ak&secret-access-key=test-sk",
			cred.BucketURL,
		)
	})

	t.Run("reject non oss url when keys are provided", func(t *testing.T) {
		cred := &entity.AliyunAuditLog{
			BucketURL:       "s3://db-auditlog-fang-test/10480461862511845599/tidb-*",
			AccessKeyID:     "test-ak",
			SecretAccessKey: "test-sk",
		}

		err := normalizeAliyunAuditLogURL(cred, "ap-southeast-1")
		require.Error(t, err)
		require.Contains(t, err.Error(), "bucket_url must start with oss:// when access keys are provided")
	})
}

func TestBuildAliyunReplayS3URL(t *testing.T) {
	tests := map[string]struct {
		bucketURL       string
		region          string
		accessKeyID     string
		secretAccessKey string
		wantURL         string
		wantErrContains string
	}{
		"success with encoded wildcard path": {
			bucketURL:       "oss://bucket/path/tidb-%2A",
			region:          "ap-southeast-1",
			accessKeyID:     "test-ak",
			secretAccessKey: "test-sk",
			wantURL:         "s3://bucket/path/tidb-?access-key=test-ak&endpoint=http%3A%2F%2Foss-ap-southeast-1.aliyuncs.com&secret-access-key=test-sk",
		},
		"reject non oss scheme": {
			bucketURL:       "s3://bucket/path/tidb-*",
			region:          "ap-southeast-1",
			accessKeyID:     "test-ak",
			secretAccessKey: "test-sk",
			wantErrContains: "bucket_url must start with oss://",
		},
		"reject empty bucket name": {
			bucketURL:       "oss://",
			region:          "ap-southeast-1",
			accessKeyID:     "test-ak",
			secretAccessKey: "test-sk",
			wantErrContains: "bucket_url does not contain bucket name",
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			got, err := buildAliyunReplayS3URL(tc.bucketURL, tc.region, tc.accessKeyID, tc.secretAccessKey)
			if tc.wantErrContains != "" {
				require.Error(t, err)
				require.Contains(t, err.Error(), tc.wantErrContains)
				return
			}

			require.NoError(t, err)
			require.Equal(t, tc.wantURL, got)
		})
	}
}

func TestReplayProviderPreparesAliyunSourceAuditCredential(t *testing.T) {
	t.Run("keep aliyun credential for non-aliyun dedicated source cluster", func(t *testing.T) {
		replayEnt := &entity.Replay{
			SourceAuditCredential: &entity.AuditLogAddr{
				Aliyun: &entity.AliyunAuditLog{
					BucketURL:       "oss://bucket/path/tidb-%2A",
					AccessKeyID:     "test-ak",
					SecretAccessKey: "test-sk",
				},
			},
		}
		sourceCluster := &mgmt.Cluster{
			ClusterId:     123,
			CloudProvider: "aws",
			Region:        "ap-southeast-1",
			DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
		}

		err := prepareReplaySourceForTest(context.Background(), &replayOperatorImpl{}, replayEnt, sourceCluster)
		require.NoError(t, err)
		require.Equal(t, "oss://bucket/path/tidb-%2A", replayEnt.SourceAuditCredential.Aliyun.BucketURL)
	})

	t.Run("keep dedicated aliyun source credential", func(t *testing.T) {
		replayEnt := &entity.Replay{
			SourceAuditCredential: &entity.AuditLogAddr{
				Aliyun: &entity.AliyunAuditLog{
					BucketURL:       "oss://bucket/path/tidb-%2A",
					AccessKeyID:     "test-ak",
					SecretAccessKey: "test-sk",
				},
			},
		}
		sourceCluster := &mgmt.Cluster{
			ClusterId:     456,
			CloudProvider: "aliyun",
			Region:        "ap-southeast-1",
			DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
		}

		err := prepareReplaySourceForTest(context.Background(), &replayOperatorImpl{}, replayEnt, sourceCluster)
		require.NoError(t, err)
		require.Equal(t, "oss://bucket/path/tidb-%2A", replayEnt.SourceAuditCredential.Aliyun.BucketURL)
	})

	t.Run("load nextgen audit setting for nextgen aliyun source cluster", func(t *testing.T) {
		mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
		operator := &replayOperatorImpl{
			MgmtSvcGrpcClient: mockCli,
		}
		replayEnt := &entity.Replay{
			SourceAuditCredential: &entity.AuditLogAddr{
				Aliyun: &entity.AliyunAuditLog{
					AccessKeyID:     "test-ak",
					SecretAccessKey: "test-sk",
				},
			},
		}
		sourceCluster := &mgmt.Cluster{
			ClusterId:     789,
			CloudProvider: "aliyun",
			Region:        "ap-southeast-1",
			DeployType:    mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &globalngv1.Cluster{
					ServicePlan: mgmt_commonv1.ServicePlan_Premium,
				},
			},
		}
		mockCli.On("GetNextGenDatabaseAuditLogSetting", mock.Anything, uint64(789)).Return(
			&globalngv1.AuditLogSetting{BucketUrl: "oss://bucket/path/tidb-%2A"},
			nil,
		).Once()

		err := prepareReplaySourceForTest(
			context.Background(),
			operator,
			replayEnt,
			sourceCluster,
		)
		require.NoError(t, err)
		require.Equal(
			t,
			"s3://bucket/path/tidb-?access-key=test-ak&endpoint=http%3A%2F%2Foss-ap-southeast-1.aliyuncs.com&secret-access-key=test-sk",
			replayEnt.SourceAuditCredential.Aliyun.BucketURL,
		)
	})

	t.Run("load nextgen audit setting url only for nextgen aliyun source cluster", func(t *testing.T) {
		mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
		operator := &replayOperatorImpl{
			MgmtSvcGrpcClient: mockCli,
		}
		replayEnt := &entity.Replay{}
		sourceCluster := &mgmt.Cluster{
			ClusterId:     791,
			CloudProvider: "aliyun",
			Region:        "ap-southeast-1",
			DeployType:    mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &globalngv1.Cluster{
					ServicePlan: mgmt_commonv1.ServicePlan_Premium,
				},
			},
		}
		mockCli.On("GetNextGenDatabaseAuditLogSetting", mock.Anything, uint64(791)).Return(
			&globalngv1.AuditLogSetting{BucketUrl: "https://audit.example.com/bucket/path/tidb-%2A"},
			nil,
		).Once()

		err := prepareReplaySourceForTest(
			context.Background(),
			operator,
			replayEnt,
			sourceCluster,
		)
		require.NoError(t, err)
		require.NotNil(t, replayEnt.SourceAuditCredential)
		require.NotNil(t, replayEnt.SourceAuditCredential.Aliyun)
		require.Equal(
			t,
			"https://audit.example.com/bucket/path/tidb-%2A",
			replayEnt.SourceAuditCredential.Aliyun.BucketURL,
		)
		require.Empty(t, replayEnt.SourceAuditCredential.Aliyun.AccessKeyID)
		require.Empty(t, replayEnt.SourceAuditCredential.Aliyun.SecretAccessKey)
	})

	t.Run("reject nextgen audit setting for nextgen gcp source cluster", func(t *testing.T) {
		mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
		operator := &replayOperatorImpl{
			MgmtSvcGrpcClient: mockCli,
		}
		replayEnt := &entity.Replay{}
		sourceCluster := &mgmt.Cluster{
			ClusterId:     793,
			CloudProvider: "gcp",
			Region:        "us-central1",
			DeployType:    mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &globalngv1.Cluster{
					ServicePlan: mgmt_commonv1.ServicePlan_Premium,
				},
			},
		}
		err := prepareReplaySourceForTest(
			context.Background(),
			operator,
			replayEnt,
			sourceCluster,
		)
		require.Error(t, err)
		require.Contains(t, err.Error(), "GKE Premium source audit log setting is not supported yet")
		require.Nil(t, replayEnt.SourceAuditCredential)
		mockCli.AssertNotCalled(t, "GetNextGenDatabaseAuditLogSetting", mock.Anything, mock.Anything)
	})

	t.Run("keep dedicated aliyun source cluster before audit log normalization", func(t *testing.T) {
		replayEnt := &entity.Replay{
			SourceAuditCredential: &entity.AuditLogAddr{
				Aliyun: &entity.AliyunAuditLog{
					AccessKeyID:     "test-ak",
					SecretAccessKey: "test-sk",
				},
			},
		}
		sourceCluster := &mgmt.Cluster{
			ClusterId:     790,
			CloudProvider: "aliyun",
			Region:        "ap-southeast-1",
			DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
		}

		err := prepareReplaySourceForTest(context.Background(), &replayOperatorImpl{}, replayEnt, sourceCluster)
		require.NoError(t, err)
		require.Empty(t, replayEnt.SourceAuditCredential.Aliyun.BucketURL)
		require.Equal(t, "test-ak", replayEnt.SourceAuditCredential.Aliyun.AccessKeyID)
		require.Equal(t, "test-sk", replayEnt.SourceAuditCredential.Aliyun.SecretAccessKey)
	})

	t.Run("reject unsupported source cluster provider", func(t *testing.T) {
		replayEnt := &entity.Replay{}
		sourceCluster := &mgmt.Cluster{
			ClusterId:     792,
			CloudProvider: "aliyun",
			Region:        "ap-southeast-1",
			DeployType:    mgmt.DeployType_DEPLOY_TYPE_UNSPECIFIED,
		}

		err := prepareReplaySourceForTest(context.Background(), &replayOperatorImpl{}, replayEnt, sourceCluster)
		require.Error(t, err)
		require.Contains(t, err.Error(), "unsupported replay cluster type")
		require.Nil(t, replayEnt.SourceAuditCredential)
	})
}

func TestPrepareReplayRequestAllowsDedicatedAliyunSource(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}

	sourceCluster := &mgmt.Cluster{
		ClusterId:      1006,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aliyun",
		Region:         "ap-southeast-1",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:      1007,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aliyun",
		Region:         "ap-southeast-1",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
				Endpoints: []*globalngv1.Endpoint{
					{
						VpcPeeringEndpoint: &globalngv1.Endpoint_VpcPeeringEndpoint{
							Host: "target-private.example.com",
							Port: 4000,
						},
					},
				},
			},
		},
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(1006)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(1007)).Return(targetCluster, nil).Once()

	replayProto, regionInfo, err := operator.prepareReplayRequest(
		context.Background(),
		&entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(9)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aliyun",
			Region:          "ap-southeast-1",
			SourceClusterID: entity.ID(1006),
			TargetClusterID: entity.ID(1007),
			SourceAuditCredential: &entity.AuditLogAddr{
				Aliyun: &entity.AliyunAuditLog{
					BucketURL: "https://audit.example.com/bucket/path/tidb-%2A",
				},
			},
		},
		entity.ReplayDesiredStateDeploy,
	)
	require.NoError(t, err)
	require.NotNil(t, replayProto)
	require.NotNil(t, regionInfo)
	require.Equal(t, "ack", replayProto.GetProvider())
	require.Equal(t, "ap-southeast-1", replayProto.GetRegion())
	require.Equal(t, "target-private.example.com:4000", replayProto.GetPremiumTargetEndpoint())
	require.Equal(t, "https://audit.example.com/bucket/path/tidb-%2A", replayProto.GetDefaults().GetSourceAuditLog().GetAliyun().GetBucketUrl())
}

func TestPrepareReplayRequestRoutesByReplayPlacementCluster(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}

	sourceCluster := &mgmt.Cluster{
		ClusterId:      1011,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
			},
		},
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:      1012,
		OrganizationId: 42,
		ProjectId:      999,
		CloudProvider:  "gcp",
		Region:         "us-central1",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(1011)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(1012)).Return(targetCluster, nil).Once()
	roleARN := "arn:aws:iam::123456789012:role/audit-log"
	mockCli.On("GetNextGenDatabaseAuditLogSetting", mock.Anything, uint64(1011)).Return(
		&globalngv1.AuditLogSetting{
			BucketUrl:         "s3://audit-bucket/path",
			CloudRoleIdentity: &roleARN,
			ExternalId:        "external-id",
		},
		nil,
	).Once()

	replayProto, regionInfo, err := operator.prepareReplayRequest(
		context.Background(),
		&entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(12)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(1011),
			TargetClusterID: entity.ID(1012),
		},
		entity.ReplayDesiredStateDeploy,
	)
	require.NoError(t, err)
	require.Equal(t, "aws", regionInfo.CloudProvider)
	require.Equal(t, "us-west-2", regionInfo.Name)
	require.Equal(t, mgmt.DeployType_DEPLOY_TYPE_NEXTGEN, regionInfo.DeployType)
	require.Equal(t, mgmt_commonv1.ServicePlan_Premium, regionInfo.ServicePlan)
	require.Equal(t, "eks", replayProto.GetProvider())
	require.Equal(t, "us-west-2", replayProto.GetRegion())
	require.Equal(t, replay_v1.Replay_CLUSTER_TYPE_PREMIUM, replayProto.GetSourceCluster().GetClusterType())
	require.Equal(t, replay_v1.Replay_CLUSTER_TYPE_DEDICATED, replayProto.GetTargetCluster().GetClusterType())
	require.Equal(t, "s3://audit-bucket/path", replayProto.GetDefaults().GetSourceAuditLog().GetAws().GetS3Url())
	require.Equal(t, "arn:aws:iam::123456789012:role/audit-log", replayProto.GetDefaults().GetSourceAuditLog().GetAws().GetRoleArn())
	require.Equal(t, "external-id", replayProto.GetDefaults().GetSourceAuditLog().GetAws().GetExternalId())
}

func TestPrepareReplayRequestAllowsTopLevelAliyunURLOnlyForDedicatedSource(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}
	sourceCluster := &mgmt.Cluster{
		ClusterId:      1101,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:      1102,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(1101)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(1102)).Return(targetCluster, nil).Once()

	replayProto, regionInfo, err := operator.prepareReplayRequest(
		context.Background(),
		&entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(11)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(1101),
			TargetClusterID: entity.ID(1102),
			SourceAuditCredential: &entity.AuditLogAddr{
				Aliyun: &entity.AliyunAuditLog{
					BucketURL: "s3://bucket/path/tidb-*?endpoint=http://oss-ap-southeast-1.aliyuncs.com",
				},
			},
		},
		entity.ReplayDesiredStateDeploy,
	)
	require.NoError(t, err)
	require.NotNil(t, replayProto)
	require.NotNil(t, regionInfo)
	require.Equal(t, "s3://bucket/path/tidb-*?endpoint=http://oss-ap-southeast-1.aliyuncs.com", replayProto.GetDefaults().GetSourceAuditLog().GetAliyun().GetBucketUrl())
}

func TestPrepareReplayRequestAllowsTopLevelAliyunKeysForDedicatedSource(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}
	sourceCluster := &mgmt.Cluster{
		ClusterId:     1103,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:     1104,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(1103)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(1104)).Return(targetCluster, nil).Once()

	replayProto, regionInfo, err := operator.prepareReplayRequest(
		context.Background(),
		&entity.Replay{
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(1103),
			TargetClusterID: entity.ID(1104),
			SourceAuditCredential: &entity.AuditLogAddr{
				Aliyun: &entity.AliyunAuditLog{
					BucketURL:       "oss://bucket/path/tidb-*",
					AccessKeyID:     "test-ak",
					SecretAccessKey: "test-sk",
				},
			},
		},
		entity.ReplayDesiredStateDeploy,
	)
	require.NoError(t, err)
	require.NotNil(t, replayProto)
	require.NotNil(t, regionInfo)
	require.Equal(t, "oss://bucket/path/tidb-*", replayProto.GetDefaults().GetSourceAuditLog().GetAliyun().GetBucketUrl())
}

func TestPrepareTargetEndpointPrefersPrivateLinkEndpoint(t *testing.T) {
	provider := &premiumReplayClusterProvider{
		targetClusterEndpoints: []*globalngv1.Endpoint{
			{
				PublicEndpoint: &globalngv1.Endpoint_PublicEndpoint{
					Host: "public.example.com",
					Port: 4001,
				},
			},
			{
				VpcPeeringEndpoint: &globalngv1.Endpoint_VpcPeeringEndpoint{
					Host: "vpc-peering.example.com",
					Port: 4002,
				},
			},
			{
				PrivateLinkEndpoints: []*globalngv1.Endpoint_PrivateLinkEndpoint{
					{
						Host: "private-link.example.com",
						Port: 4003,
					},
				},
			},
		},
	}

	targetEndpoint, err := provider.PrepareTargetEndpoint()
	require.NoError(t, err)
	require.Equal(t, "private-link.example.com:4003", targetEndpoint)
}

func TestPrepareTargetEndpointIgnoresPublicEndpointWhenVpcPeeringExists(t *testing.T) {
	provider := &premiumReplayClusterProvider{
		targetClusterEndpoints: []*globalngv1.Endpoint{
			{
				PublicEndpoint: &globalngv1.Endpoint_PublicEndpoint{
					Host: "public.example.com",
					Port: 4001,
				},
			},
			{
				VpcPeeringEndpoint: &globalngv1.Endpoint_VpcPeeringEndpoint{
					Host: "vpc-peering.example.com",
					Port: 4002,
				},
			},
		},
	}

	targetEndpoint, err := provider.PrepareTargetEndpoint()
	require.NoError(t, err)
	require.Equal(t, "vpc-peering.example.com:4002", targetEndpoint)
}

func TestPrepareTargetEndpointRejectsPublicEndpoint(t *testing.T) {
	provider := &premiumReplayClusterProvider{
		targetClusterEndpoints: []*globalngv1.Endpoint{
			{
				PublicEndpoint: &globalngv1.Endpoint_PublicEndpoint{
					Host: "public.example.com",
					Port: 4001,
				},
			},
		},
	}

	targetEndpoint, err := provider.PrepareTargetEndpoint()
	require.Error(t, err)
	require.Empty(t, targetEndpoint)
	require.Contains(t, err.Error(), "has no usable nextgen private endpoint")
}

func TestPremiumTargetEndpointUsesTargetClusterEndpoints(t *testing.T) {
	provider := newPremiumReplayClusterProvider(nil, &mgmt.Cluster{
		ClusterId:  1003,
		DeployType: mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
				Endpoints: []*globalngv1.Endpoint{
					{
						VpcPeeringEndpoint: &globalngv1.Endpoint_VpcPeeringEndpoint{
							Host: "target-private.example.com",
							Port: 4000,
						},
					},
				},
			},
		},
	}, &entity.Replay{})

	targetEndpoint, err := provider.PrepareTargetEndpoint()
	require.NoError(t, err)
	require.Equal(t, "target-private.example.com:4000", targetEndpoint)
}

func TestPrepareReplayRequestPremiumTargetRequiresEndpointForAutoTasks(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}

	sourceCluster := &mgmt.Cluster{
		ClusterId:     1201,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:     1202,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
			},
		},
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(1201)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(1202)).Return(targetCluster, nil).Once()

	_, _, err := operator.prepareReplayRequest(
		context.Background(),
		&entity.Replay{
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(1201),
			TargetClusterID: entity.ID(1202),
		},
		entity.ReplayDesiredStateDeploy,
	)
	require.Error(t, err)
	require.Contains(t, err.Error(), "missing nextgen cluster endpoints")
}

func TestPrepareReplayRequestDedicatedToPremiumEKSAutoTasks(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}

	sourceCluster := &mgmt.Cluster{
		ClusterId:      1211,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:      1212,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
				Endpoints: []*globalngv1.Endpoint{
					{
						VpcPeeringEndpoint: &globalngv1.Endpoint_VpcPeeringEndpoint{
							Host: "target-private.example.com",
							Port: 4000,
						},
					},
				},
			},
		},
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(1211)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(1212)).Return(targetCluster, nil).Once()

	replayProto, regionInfo, err := operator.prepareReplayRequest(
		context.Background(),
		&entity.Replay{
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(1211),
			TargetClusterID: entity.ID(1212),
		},
		entity.ReplayDesiredStateDeploy,
	)
	require.NoError(t, err)
	require.NotNil(t, replayProto)
	require.NotNil(t, regionInfo)
	require.Equal(t, "eks", replayProto.GetProvider())
	require.Equal(t, "us-west-2", replayProto.GetRegion())
	require.Equal(t, "target-private.example.com:4000", replayProto.GetPremiumTargetEndpoint())
	require.Nil(t, replayProto.GetDefaults().GetSourceAuditLog())
	require.Equal(t, replay_v1.Replay_CLUSTER_TYPE_DEDICATED, replayProto.GetSourceCluster().GetClusterType())
	require.Equal(t, "eks", replayProto.GetSourceCluster().GetProvider())
	require.Equal(t, replay_v1.Replay_CLUSTER_TYPE_PREMIUM, replayProto.GetTargetCluster().GetClusterType())
	require.Equal(t, "eks", replayProto.GetTargetCluster().GetProvider())
	require.Equal(t, "aws", regionInfo.CloudProvider)
	require.Equal(t, "us-west-2", regionInfo.Name)
	require.Equal(t, mgmt.DeployType_DEPLOY_TYPE_NEXTGEN, regionInfo.DeployType)
	require.Equal(t, mgmt_commonv1.ServicePlan_Premium, regionInfo.ServicePlan)
}

func TestPrepareReplayRequestRejectsUnsupportedTargetClusterProvider(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}

	sourceCluster := &mgmt.Cluster{
		ClusterId:     1205,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:     1206,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Essential,
			},
		},
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(1205)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(1206)).Return(targetCluster, nil).Once()

	replayProto, regionInfo, err := operator.prepareReplayRequest(
		context.Background(),
		&entity.Replay{
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(1205),
			TargetClusterID: entity.ID(1206),
		},
		entity.ReplayDesiredStateDeploy,
	)
	require.Nil(t, replayProto)
	require.Nil(t, regionInfo)
	require.Error(t, err)
	require.Contains(t, err.Error(), "resolve target replay provider")
	require.Contains(t, err.Error(), "unsupported replay cluster type")
}

func TestPrepareReplayRequestAllowsMissingSourceClusterIDForTargetPlacement(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}

	targetCluster := &mgmt.Cluster{
		ClusterId:      1207,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(1207)).Return(targetCluster, nil).Once()

	replayProto, regionInfo, err := operator.prepareReplayRequest(
		context.Background(),
		&entity.Replay{
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			TargetClusterID: entity.ID(1207),
		},
		entity.ReplayDesiredStateDeploy,
	)
	require.NoError(t, err)
	require.Nil(t, replayProto.GetSourceCluster())
	require.Equal(t, "1207", replayProto.GetTargetCluster().GetClusterId())
	require.Equal(t, "aws", regionInfo.CloudProvider)
	require.Equal(t, "us-west-2", regionInfo.Name)
	mockCli.AssertNumberOfCalls(t, "GetClusterByID", 1)
}

func TestPrepareReplayRequestPremiumSourceLoadsAuditLogFromMgmt(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}

	sourceCluster := &mgmt.Cluster{
		ClusterId:     1203,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
			},
		},
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:     1204,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
				Endpoints: []*globalngv1.Endpoint{
					{
						VpcPeeringEndpoint: &globalngv1.Endpoint_VpcPeeringEndpoint{
							Host: "target-private.example.com",
							Port: 4000,
						},
					},
				},
			},
		},
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(1203)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(1204)).Return(targetCluster, nil).Once()
	roleARN := "arn:aws:iam::123456789012:role/replay-audit"
	mockCli.On("GetNextGenDatabaseAuditLogSetting", mock.Anything, uint64(1203)).Return(
		&globalngv1.AuditLogSetting{
			BucketUrl:         "s3://premium-source-audit/path",
			CloudRoleIdentity: &roleARN,
			ExternalId:        "premium-external-id",
		},
		nil,
	).Once()

	replayProto, _, err := operator.prepareReplayRequest(
		context.Background(),
		&entity.Replay{
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(1203),
			TargetClusterID: entity.ID(1204),
		},
		entity.ReplayDesiredStateDeploy,
	)
	require.NoError(t, err)
	require.Equal(t, "target-private.example.com:4000", replayProto.GetPremiumTargetEndpoint())
	require.NotNil(t, replayProto.GetDefaults().GetSourceAuditLog())
	require.Equal(t, "s3://premium-source-audit/path", replayProto.GetDefaults().GetSourceAuditLog().GetAws().GetS3Url())
	require.Equal(t, "arn:aws:iam::123456789012:role/replay-audit", replayProto.GetDefaults().GetSourceAuditLog().GetAws().GetRoleArn())
	require.Equal(t, "premium-external-id", replayProto.GetDefaults().GetSourceAuditLog().GetAws().GetExternalId())
}

func TestPrepareReplayRequestSkipsTopLevelAliyunForExplicitTasks(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}
	sourceCluster := &mgmt.Cluster{
		ClusterId:     101,
		CloudProvider: "aliyun",
		Region:        "ap-southeast-1",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
			},
		},
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:     202,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(101)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(202)).Return(targetCluster, nil).Once()

	replayEnt := &entity.Replay{
		CloudProvider:   "aliyun",
		Region:          "ap-southeast-1",
		SourceClusterID: entity.ID(101),
		TargetClusterID: entity.ID(202),
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:       "oss://top-level/path/tidb-*",
				AccessKeyID:     "enc:top-level-ak",
				SecretAccessKey: "enc:top-level-sk",
			},
		},
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "task-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       "oss://bucket/path/tidb-%2A",
								AccessKeyID:     "task-ak",
								SecretAccessKey: "task-sk",
							},
						},
					},
				},
				DataTarget: entity.ReplayDataTarget{Endpoint: "127.0.0.1:4000"},
			},
		},
	}

	replayProto, _, err := operator.prepareReplayRequest(context.Background(), replayEnt, entity.ReplayDesiredStateDeploy)
	require.NoError(t, err)
	require.Len(t, replayProto.Tasks, 1)
	require.NotNil(t, replayProto.Tasks[0].Source)
	require.Len(t, replayProto.Tasks[0].Source.AuditLogs, 1)
	require.Nil(t, replayProto.GetDefaults().GetSourceAuditLog())
	gotAliyun := replayProto.Tasks[0].Source.AuditLogs[0].GetAliyun()
	require.NotNil(t, gotAliyun)
	require.Contains(t, gotAliyun.BucketUrl, "s3://bucket/path/tidb-")
	require.Equal(t, "oss://top-level/path/tidb-*", replayEnt.SourceAuditCredential.Aliyun.BucketURL)
	require.Equal(t, "enc:top-level-ak", replayEnt.SourceAuditCredential.Aliyun.AccessKeyID)
	require.Equal(t, "enc:top-level-sk", replayEnt.SourceAuditCredential.Aliyun.SecretAccessKey)
	require.Equal(
		t,
		"oss://bucket/path/tidb-%2A",
		replayEnt.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.BucketURL,
	)
	require.Equal(t, "task-ak", replayEnt.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.AccessKeyID)
	require.Equal(t, "task-sk", replayEnt.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.SecretAccessKey)
	mockCli.AssertNotCalled(t, "GetNextGenDatabaseAuditLogSetting", mock.Anything, mock.Anything)
}

func TestPrepareReplayRequestAllowsTaskLevelAliyunS3URLForDedicatedSource(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}
	sourceCluster := &mgmt.Cluster{
		ClusterId:     301,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:     302,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(301)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(302)).Return(targetCluster, nil).Once()

	replayEnt := &entity.Replay{
		CloudProvider:   "aws",
		Region:          "us-west-2",
		SourceClusterID: entity.ID(301),
		TargetClusterID: entity.ID(302),
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "task-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       "s3://bucket/path/tidb-*?endpoint=http://oss-ap-southeast-1.aliyuncs.com&access-key=old-ak&secret-access-key=old-sk",
								AccessKeyID:     "new-ak",
								SecretAccessKey: "new-sk",
							},
						},
					},
				},
				DataTarget: entity.ReplayDataTarget{Endpoint: "127.0.0.1:4000"},
			},
		},
	}

	replayProto, regionInfo, err := operator.prepareReplayRequest(context.Background(), replayEnt, entity.ReplayDesiredStateDeploy)
	require.NoError(t, err)
	require.NotNil(t, replayProto)
	require.NotNil(t, regionInfo)
	require.Len(t, replayProto.Tasks, 1)
	require.Len(t, replayProto.Tasks[0].Source.AuditLogs, 1)
	require.Equal(
		t,
		"s3://bucket/path/tidb-*?endpoint=http://oss-ap-southeast-1.aliyuncs.com&access-key=old-ak&secret-access-key=old-sk",
		replayProto.Tasks[0].Source.AuditLogs[0].GetAliyun().GetBucketUrl(),
	)
}

func TestPrepareReplayRequestClonesAndDecryptsWithoutMutatingCaller(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}
	sourceCluster := &mgmt.Cluster{
		ClusterId:     400,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:     401,
		CloudProvider: "gcp",
		Region:        "us-central1",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(400)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(401)).Return(targetCluster, nil).Once()

	replayEnt := &entity.Replay{
		CloudProvider:   "aws",
		Region:          "us-west-2",
		SourceClusterID: entity.ID(400),
		TargetClusterID: entity.ID(401),
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:       encryptedAliyunBucketURLForTest("oss://bucket/source-path"),
				AccessKeyID:     "enc:test-ak",
				SecretAccessKey: "enc:test-sk",
			},
		},
		ReplayTasks: entity.ReplayTaskList{
			{
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       encryptedAliyunBucketURLForTest("s3://bucket/task-path"),
								AccessKeyID:     "enc:task-ak",
								SecretAccessKey: "enc:task-sk",
							},
						},
					},
				},
			},
		},
	}

	replayProto, regionInfo, err := operator.prepareReplayRequest(context.Background(), replayEnt, entity.ReplayDesiredStateDeploy)
	require.NoError(t, err)
	require.NotNil(t, replayProto)
	require.NotNil(t, regionInfo)
	require.Equal(t, "aws", replayEnt.CloudProvider)
	require.Equal(t, "us-west-2", replayEnt.Region)
	require.Equal(t, encryptedAliyunBucketURLForTest("oss://bucket/source-path"), replayEnt.SourceAuditCredential.Aliyun.BucketURL)
	require.Equal(t, "enc:test-ak", replayEnt.SourceAuditCredential.Aliyun.AccessKeyID)
	require.Equal(t, "enc:test-sk", replayEnt.SourceAuditCredential.Aliyun.SecretAccessKey)
	require.Equal(
		t,
		encryptedAliyunBucketURLForTest("s3://bucket/task-path"),
		replayEnt.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.BucketURL,
	)
	require.Equal(t, "enc:task-ak", replayEnt.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.AccessKeyID)
	require.Equal(t, "enc:task-sk", replayEnt.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.SecretAccessKey)
	require.Nil(t, replayProto.GetDefaults().GetSourceAuditLog())
	require.Equal(t, "s3://bucket/task-path", replayProto.GetTasks()[0].GetSource().GetAuditLogs()[0].GetAliyun().GetBucketUrl())
}

func TestModifyRoutesAndBuildsProtoFromResolvedTargetCluster(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	targetCluster := &mgmt.Cluster{
		ClusterId:      902,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "gcp",
		Region:         "us-central1",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	sourceCluster := &mgmt.Cluster{
		ClusterId:     901,
		CloudProvider: "aws",
		Region:        "us-west-2",
		DeployType:    mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(901)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(902)).Return(targetCluster, nil)

	replayCli.On(
		"UpdateReplay",
		mock.Anything,
		mock.MatchedBy(func(region *regionalapi.Region) bool {
			return region != nil &&
				region.TenantID == 42 &&
				region.CloudProvider == "gcp" &&
				region.Name == "us-central1"
		}),
		mock.MatchedBy(func(req *replay_v1.UpdateReplayRequest) bool {
			replay := req.GetReplay()
			return replay != nil &&
				replay.GetProvider() == "gke" &&
				replay.GetRegion() == "us-central1" &&
				replay.GetTargetCluster().GetClusterId() == "902" &&
				replay.GetTargetCluster().GetProvider() == "gke"
		}),
	).Return(&replay_v1.Replay{}, nil).Once()

	err := operator.Modify(context.Background(), &domain.ModifyReplayReq{
		Replay: &entity.Replay{
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "gcp",
			Region:          "us-central1",
			SourceClusterID: entity.ID(901),
			TargetClusterID: entity.ID(902),
		},
		State: entity.ReplayDesiredStateDeploy,
	})
	require.NoError(t, err)
}

func TestPrepareReplayRequestSourceRouteKeepsTargetClusterRef(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}

	sourceCluster := &mgmt.Cluster{
		ClusterId:      910,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:      911,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "gcp",
		Region:         "us-central1",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(910)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(911)).Return(targetCluster, nil).Once()

	replayProto, regionInfo, err := operator.prepareReplayRequest(
		context.Background(),
		&entity.Replay{
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(910),
			TargetClusterID: entity.ID(911),
		},
		entity.ReplayDesiredStateDeploy,
	)
	require.NoError(t, err)
	require.Equal(t, "eks", replayProto.GetProvider())
	require.Equal(t, "us-west-2", replayProto.GetRegion())
	require.Equal(t, "gke", replayProto.GetTargetCluster().GetProvider())
	require.Equal(t, "us-central1", replayProto.GetTargetCluster().GetRegion())
	require.Equal(t, "aws", regionInfo.CloudProvider)
	require.Equal(t, "us-west-2", regionInfo.Name)
}

func TestGetStatusRoutesWithResolvedPremiumTargetCluster(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	targetCluster := &mgmt.Cluster{
		ClusterId:      903,
		OrganizationId: 42,
		CloudProvider:  "ack",
		Region:         "ap-southeast-1",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
			},
		},
	}
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(903)).Return(targetCluster, nil).Once()

	replayCli.On(
		"GetReplay",
		mock.Anything,
		mock.MatchedBy(func(region *regionalapi.Region) bool {
			return region != nil &&
				region.TenantID == 42 &&
				region.CloudProvider == "ack" &&
				region.Name == "ap-southeast-1" &&
				region.DeployType == mgmt.DeployType_DEPLOY_TYPE_NEXTGEN &&
				region.ServicePlan == mgmt_commonv1.ServicePlan_Premium
		}),
		mock.MatchedBy(func(req *replay_v1.GetReplayRequest) bool {
			return req.GetReplayId() == "replay-7"
		}),
	).Return(&replay_v1.GetReplayResponse{
		Replay: &replay_v1.Replay{
			State: replay_v1.Replay_STATE_ACTIVE,
		},
		Status: &replay_v1.ReplayStatus{
			Phase: replay_v1.ReplayStatus_PHASE_RUNNING,
		},
	}, nil).Once()

	resp, err := operator.GetStatus(context.Background(), &domain.GetReplayStatusReq{
		Replay: &entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(7)}},
			TenantID:        entity.ID(42),
			CloudProvider:   "ack",
			Region:          "ap-southeast-1",
			TargetClusterID: entity.ID(903),
		},
	})
	require.NoError(t, err)
	require.Equal(t, entity.ReplayPhaseRunning, resp.Phase)
	require.Equal(t, entity.ReplayDesiredStateActive, resp.State)
}

func TestGetStatusRoutesWithResolvedSourcePlacementCluster(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	sourceCluster := &mgmt.Cluster{
		ClusterId:      9041,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aliyun",
		Region:         "ap-southeast-1",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
			},
		},
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:      9042,
		OrganizationId: 42,
		ProjectId:      999,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9042)).Return(targetCluster, nil).Once()
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9041)).Return(sourceCluster, nil).Once()

	replayCli.On(
		"GetReplay",
		mock.Anything,
		mock.MatchedBy(func(region *regionalapi.Region) bool {
			return region != nil &&
				region.TenantID == 42 &&
				region.CloudProvider == "aliyun" &&
				region.Name == "ap-southeast-1" &&
				region.DeployType == mgmt.DeployType_DEPLOY_TYPE_NEXTGEN &&
				region.ServicePlan == mgmt_commonv1.ServicePlan_Premium
		}),
		mock.MatchedBy(func(req *replay_v1.GetReplayRequest) bool {
			return req.GetReplayId() == "replay-12"
		}),
	).Return(&replay_v1.GetReplayResponse{
		Replay: &replay_v1.Replay{
			State: replay_v1.Replay_STATE_ACTIVE,
		},
		Status: &replay_v1.ReplayStatus{
			Phase: replay_v1.ReplayStatus_PHASE_RUNNING,
		},
	}, nil).Once()

	resp, err := operator.GetStatus(context.Background(), &domain.GetReplayStatusReq{
		Replay: &entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(12)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aliyun",
			Region:          "ap-southeast-1",
			SourceClusterID: entity.ID(9041),
			TargetClusterID: entity.ID(9042),
		},
	})
	require.NoError(t, err)
	require.Equal(t, entity.ReplayPhaseRunning, resp.Phase)
	require.Equal(t, entity.ReplayDesiredStateActive, resp.State)
}

func TestGetStatusRoutesUnsupportedTargetPlacementCluster(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	targetCluster := &mgmt.Cluster{
		ClusterId:      9047,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Essential,
			},
		},
	}
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9047)).Return(targetCluster, nil).Once()

	replayCli.On(
		"GetReplay",
		mock.Anything,
		mock.MatchedBy(func(region *regionalapi.Region) bool {
			return region != nil &&
				region.TenantID == 42 &&
				region.CloudProvider == "aws" &&
				region.Name == "us-west-2" &&
				region.DeployType == mgmt.DeployType_DEPLOY_TYPE_NEXTGEN &&
				region.ServicePlan == mgmt_commonv1.ServicePlan_Essential
		}),
		mock.MatchedBy(func(req *replay_v1.GetReplayRequest) bool {
			return req.GetReplayId() == "replay-15"
		}),
	).Return(&replay_v1.GetReplayResponse{
		Replay: &replay_v1.Replay{
			State: replay_v1.Replay_STATE_ACTIVE,
		},
		Status: &replay_v1.ReplayStatus{
			Phase: replay_v1.ReplayStatus_PHASE_RUNNING,
		},
	}, nil).Once()

	resp, err := operator.GetStatus(context.Background(), &domain.GetReplayStatusReq{
		Replay: &entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(15)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			TargetClusterID: entity.ID(9047),
		},
	})
	require.NoError(t, err)
	require.Equal(t, entity.ReplayPhaseRunning, resp.Phase)
	require.Equal(t, entity.ReplayDesiredStateActive, resp.State)
}

func TestStartRoutesUnsupportedSourcePlacementCluster(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	sourceCluster := &mgmt.Cluster{
		ClusterId:      9048,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Starter,
			},
		},
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:      9049,
		OrganizationId: 42,
		ProjectId:      999,
		CloudProvider:  "gcp",
		Region:         "us-central1",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9049)).Return(targetCluster, nil).Once()
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9048)).Return(sourceCluster, nil).Once()

	replayCli.On(
		"StartReplay",
		mock.Anything,
		mock.MatchedBy(func(region *regionalapi.Region) bool {
			return region != nil &&
				region.TenantID == 42 &&
				region.CloudProvider == "aws" &&
				region.Name == "us-west-2" &&
				region.DeployType == mgmt.DeployType_DEPLOY_TYPE_NEXTGEN &&
				region.ServicePlan == mgmt_commonv1.ServicePlan_Starter
		}),
		mock.MatchedBy(func(req *replay_v1.StartReplayRequest) bool {
			return req.GetReplayId() == "replay-16"
		}),
	).Return(&replay_v1.Replay{}, nil).Once()

	err := operator.Start(context.Background(), &domain.StartReplayReq{
		Replay: &entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(16)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(9048),
			TargetClusterID: entity.ID(9049),
		},
	})
	require.NoError(t, err)
}

func TestGetStatusReturnsTargetRouteCandidateLookupError(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9044)).Return(
		(*mgmt.Cluster)(nil),
		bizerr.New(bizerr.ErrCodeUnknown, "target mgmt timeout"),
	).Once()

	resp, err := operator.GetStatus(context.Background(), &domain.GetReplayStatusReq{
		Replay: &entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(13)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(9043),
			TargetClusterID: entity.ID(9044),
		},
	})
	require.Nil(t, resp)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeUnknown, err.RawErrCode())
	require.Contains(t, err.Error(), "get replay route cluster 9044 info")
	require.Contains(t, err.Error(), "target mgmt timeout")
	mockCli.AssertNumberOfCalls(t, "GetAnyClusterByID", 1)
	replayCli.AssertNumberOfCalls(t, "GetReplay", 0)
}

func TestGetStatusReturnsSourceLookupErrorBeforeTargetFallback(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	targetCluster := &mgmt.Cluster{
		ClusterId:      9046,
		OrganizationId: 42,
		ProjectId:      999,
		CloudProvider:  "gcp",
		Region:         "us-central1",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9046)).Return(targetCluster, nil).Once()
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9045)).Return(
		(*mgmt.Cluster)(nil),
		bizerr.New(bizerr.ErrCodeUnknown, "source mgmt timeout"),
	).Once()

	resp, err := operator.GetStatus(context.Background(), &domain.GetReplayStatusReq{
		Replay: &entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(14)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(9045),
			TargetClusterID: entity.ID(9046),
		},
	})
	require.Nil(t, resp)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeUnknown, err.RawErrCode())
	require.Contains(t, err.Error(), "get replay route cluster 9045 info")
	require.Contains(t, err.Error(), "source mgmt timeout")
	mockCli.AssertNumberOfCalls(t, "GetAnyClusterByID", 2)
	replayCli.AssertNumberOfCalls(t, "GetReplay", 0)
}

func TestGetStatusRoutesSourcePlacementWhenTargetNotFound(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(905)).Return(
		(*mgmt.Cluster)(nil),
		bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster 905 is not found"),
	).Once()

	sourceCluster := &mgmt.Cluster{
		ClusterId:      906,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
			},
		},
	}
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(906)).Return(sourceCluster, nil).Once()

	replayCli.On(
		"GetReplay",
		mock.Anything,
		mock.MatchedBy(func(region *regionalapi.Region) bool {
			return region != nil &&
				region.TenantID == 42 &&
				region.CloudProvider == "aws" &&
				region.Name == "us-west-2" &&
				region.DeployType == mgmt.DeployType_DEPLOY_TYPE_NEXTGEN &&
				region.ServicePlan == mgmt_commonv1.ServicePlan_Premium
		}),
		mock.MatchedBy(func(req *replay_v1.GetReplayRequest) bool {
			return req.GetReplayId() == "replay-10"
		}),
	).Return(&replay_v1.GetReplayResponse{
		Replay: &replay_v1.Replay{
			State: replay_v1.Replay_STATE_ACTIVE,
		},
		Status: &replay_v1.ReplayStatus{
			Phase: replay_v1.ReplayStatus_PHASE_RUNNING,
		},
	}, nil).Once()

	replayEnt := &entity.Replay{
		Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(10)}},
		TenantID:        entity.ID(42),
		ProjectID:       entity.ID(84),
		CloudProvider:   "aws",
		Region:          "us-west-2",
		SourceClusterID: entity.ID(906),
		TargetClusterID: entity.ID(905),
		ReplayStatus:    &entity.ReplayStatus{Phase: entity.ReplayPhasePending},
	}
	resp, err := operator.GetStatus(context.Background(), &domain.GetReplayStatusReq{Replay: replayEnt})
	require.NoError(t, err)
	require.Equal(t, entity.ReplayPhaseRunning, resp.Phase)
	require.Equal(t, entity.ReplayDesiredStateActive, resp.State)
	mockCli.AssertNumberOfCalls(t, "GetAnyClusterByID", 2)
}

func TestGetStatusReturnsPlacementMismatchWhenRouteCandidatesNotFound(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(908)).Return(
		(*mgmt.Cluster)(nil),
		bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster 908 is not found"),
	).Once()
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(907)).Return(
		(*mgmt.Cluster)(nil),
		bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster 907 is not found"),
	).Once()

	resp, err := operator.GetStatus(context.Background(), &domain.GetReplayStatusReq{
		Replay: &entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(17)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(907),
			TargetClusterID: entity.ID(908),
			ReplayStatus:    &entity.ReplayStatus{Phase: entity.ReplayPhasePending},
		},
	})
	require.Nil(t, resp)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
	require.Contains(t, err.Error(), "replay deployment placement does not match source or target cluster")
	mockCli.AssertNumberOfCalls(t, "GetAnyClusterByID", 2)
	replayCli.AssertNumberOfCalls(t, "GetReplay", 0)
}

func TestGetStatusReturnsTargetClusterLookupError(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	lookupErr := bizerr.New(bizerr.ErrCodeUnknown, "mgmt timeout")
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(906)).Return((*mgmt.Cluster)(nil), lookupErr).Once()

	resp, err := operator.GetStatus(context.Background(), &domain.GetReplayStatusReq{
		Replay: &entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(11)}},
			TenantID:        entity.ID(42),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			TargetClusterID: entity.ID(906),
			ReplayStatus:    &entity.ReplayStatus{Phase: entity.ReplayPhasePending},
		},
	})
	require.Nil(t, resp)
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeUnknown, err.RawErrCode())
	require.Contains(t, err.Error(), "get replay route cluster 906 info")
	require.Contains(t, err.Error(), "mgmt timeout")
	replayCli.AssertNumberOfCalls(t, "GetReplay", 0)
}

func TestDestroyRoutesWithResolvedPremiumTargetCluster(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	targetCluster := &mgmt.Cluster{
		ClusterId:      9101,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
			},
		},
	}
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9101)).Return(targetCluster, nil).Once()

	replayCli.On(
		"DeleteReplay",
		mock.Anything,
		mock.MatchedBy(func(region *regionalapi.Region) bool {
			return region != nil &&
				region.TenantID == 42 &&
				region.CloudProvider == "aws" &&
				region.Name == "us-west-2" &&
				region.DeployType == mgmt.DeployType_DEPLOY_TYPE_NEXTGEN &&
				region.ServicePlan == mgmt_commonv1.ServicePlan_Premium
		}),
		mock.MatchedBy(func(req *replay_v1.DeleteReplayRequest) bool {
			return req.GetReplayId() == "replay-18"
		}),
	).Return(&replay_v1.Replay{}, nil).Once()

	err := operator.Destroy(context.Background(), &domain.DestroyReplayReq{
		Replay: &entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(18)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			TargetClusterID: entity.ID(9101),
			ReplayStatus:    &entity.ReplayStatus{Phase: entity.ReplayPhaseDeleting},
		},
	})
	require.NoError(t, err)
	mockCli.AssertNumberOfCalls(t, "GetAnyClusterByID", 1)
}

func TestDestroyReturnsPlacementMismatchWhenTargetRouteCandidateNotFound(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9105)).Return(
		(*mgmt.Cluster)(nil),
		bizerr.New(bizerr.ErrCodeRecordNotFound, "cluster 9105 is not found"),
	).Once()

	err := operator.Destroy(context.Background(), &domain.DestroyReplayReq{
		Replay: &entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(22)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			TargetClusterID: entity.ID(9105),
			ReplayStatus:    &entity.ReplayStatus{Phase: entity.ReplayPhaseDeleting},
		},
	})
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
	require.Contains(t, err.Error(), "build replay route region")
	require.Contains(t, err.Error(), "replay deployment placement does not match source or target cluster")
	replayCli.AssertNumberOfCalls(t, "DeleteReplay", 0)
}

func TestDestroyDoesNotFallbackOnRouteCandidateLookupError(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9102)).Return(
		(*mgmt.Cluster)(nil),
		bizerr.New(bizerr.ErrCodeUnknown, "mgmt timeout"),
	).Once()

	err := operator.Destroy(context.Background(), &domain.DestroyReplayReq{
		Replay: &entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(19)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			TargetClusterID: entity.ID(9102),
			ReplayStatus:    &entity.ReplayStatus{Phase: entity.ReplayPhaseDeleting},
		},
	})
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeUnknown, err.RawErrCode())
	require.Contains(t, err.Error(), "build replay route region")
	require.Contains(t, err.Error(), "get replay route cluster 9102 info")
	require.Contains(t, err.Error(), "mgmt timeout")
	replayCli.AssertNumberOfCalls(t, "DeleteReplay", 0)
}

func TestDestroyDoesNotFallbackWhenRouteCandidatesExistButPlacementMismatch(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	replayCli := regionalapi_mock.NewReplayClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient:       mockCli,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               replayTestEncryptor{},
	}

	targetCluster := &mgmt.Cluster{
		ClusterId:      9103,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "gcp",
		Region:         "us-central1",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	sourceCluster := &mgmt.Cluster{
		ClusterId:      9104,
		OrganizationId: 42,
		ProjectId:      999,
		CloudProvider:  "aws",
		Region:         "us-west-2",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
	}
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9103)).Return(targetCluster, nil).Once()
	mockCli.On("GetAnyClusterByID", mock.Anything, uint64(9104)).Return(sourceCluster, nil).Once()

	err := operator.Destroy(context.Background(), &domain.DestroyReplayReq{
		Replay: &entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(20)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "aws",
			Region:          "us-west-2",
			SourceClusterID: entity.ID(9104),
			TargetClusterID: entity.ID(9103),
			ReplayStatus:    &entity.ReplayStatus{Phase: entity.ReplayPhaseDeleting},
		},
	})
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
	require.Contains(t, err.Error(), "replay deployment placement does not match source or target cluster")
	replayCli.AssertNumberOfCalls(t, "DeleteReplay", 0)
}

func TestPrepareReplayRequestSupportsPremiumToPremiumReplay(t *testing.T) {
	mockCli := grpcli_mock.NewMgmtSvcGrpcClient(t)
	operator := &replayOperatorImpl{
		MgmtSvcGrpcClient: mockCli,
		encryptor:         replayTestEncryptor{},
	}

	sourceCluster := &mgmt.Cluster{
		ClusterId:      1003,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "alicloud",
		Region:         "ap-southeast-1",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
			},
		},
	}
	targetCluster := &mgmt.Cluster{
		ClusterId:      1004,
		OrganizationId: 42,
		ProjectId:      84,
		CloudProvider:  "alicloud",
		Region:         "ap-southeast-1",
		DeployType:     mgmt.DeployType_DEPLOY_TYPE_NEXTGEN,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &globalngv1.Cluster{
				ServicePlan: mgmt_commonv1.ServicePlan_Premium,
				Endpoints: []*globalngv1.Endpoint{
					{
						VpcPeeringEndpoint: &globalngv1.Endpoint_VpcPeeringEndpoint{
							Host: "target-private.example.com",
							Port: 4000,
						},
					},
				},
			},
		},
	}
	mockCli.On("GetClusterByID", mock.Anything, uint64(1003)).Return(sourceCluster, nil).Once()
	mockCli.On("GetClusterByID", mock.Anything, uint64(1004)).Return(targetCluster, nil).Once()
	mockCli.On("GetNextGenDatabaseAuditLogSetting", mock.Anything, uint64(1003)).Return(
		&globalngv1.AuditLogSetting{BucketUrl: "oss://bucket/path/tidb-%2A"},
		nil,
	).Once()

	replayProto, regionInfo, err := operator.prepareReplayRequest(
		context.Background(),
		&entity.Replay{
			Updatable:       entity.Updatable{Base: entity.Base{ID: entity.ID(8)}},
			TenantID:        entity.ID(42),
			ProjectID:       entity.ID(84),
			CloudProvider:   "alicloud",
			Region:          "ap-southeast-1",
			SourceClusterID: entity.ID(1003),
			TargetClusterID: entity.ID(1004),
			SourceAuditCredential: &entity.AuditLogAddr{
				Aliyun: &entity.AliyunAuditLog{
					BucketURL:       "oss://bucket/path/tidb-%2A",
					AccessKeyID:     "test-ak",
					SecretAccessKey: "test-sk",
				},
			},
		},
		entity.ReplayDesiredStateDeploy,
	)
	require.NoError(t, err)
	require.Equal(t, "alicloud", regionInfo.CloudProvider)
	require.Equal(t, "ap-southeast-1", regionInfo.Name)
	require.Equal(t, mgmt.DeployType_DEPLOY_TYPE_NEXTGEN, regionInfo.DeployType)
	require.Equal(t, mgmt_commonv1.ServicePlan_Premium, regionInfo.ServicePlan)
	require.Equal(t, replay_v1.Replay_CLUSTER_TYPE_PREMIUM, replayProto.GetSourceCluster().GetClusterType())
	require.Equal(t, replay_v1.Replay_CLUSTER_TYPE_PREMIUM, replayProto.GetTargetCluster().GetClusterType())
	require.Equal(t, "ack", replayProto.GetProvider())
	require.Equal(t, "ack", replayProto.GetSourceCluster().GetProvider())
	require.Equal(t, "ack", replayProto.GetTargetCluster().GetProvider())
	require.Equal(t, "target-private.example.com:4000", replayProto.GetPremiumTargetEndpoint())
	require.NotNil(t, replayProto.GetDefaults().GetSourceAuditLog())
	require.Equal(
		t,
		"s3://bucket/path/tidb-?access-key=test-ak&endpoint=http%3A%2F%2Foss-ap-southeast-1.aliyuncs.com&secret-access-key=test-sk",
		replayProto.GetDefaults().GetSourceAuditLog().GetAliyun().GetBucketUrl(),
	)
}