package server

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"testing"

	dataplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"

	config2 "github.com/tidbcloud/cluster-controller/server/pkg/config"
	"github.com/tidbcloud/cluster-controller/server/server/cvt"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
)

const (
	replayNewAWSRoleARN         = "arn:aws:iam::123456789012:role/replay-new"
	replayNewExternalID         = "external-new"
	replayUpdatedExternalID     = "new-external-id"
	replayNewRootUsername       = "new-root"
	replayOldSourceClusterID    = "source-old"
	replayOldPrefixDefaultS3URL = "s3://source-audit/old-prefix/default_tidb-"
	replaySourceDefaultS3URL    = "s3://source-audit/prefix/default_tidb-"
	replayNewDataplaneName      = "dp-new"
	replayTaskGroup0Name        = "group-0"
	replayTaskGroupDefaultName  = "group-default"
	replayNewKubeConfig         = "new-kubeconfig"
	replayPremiumTestKubeconfig = `apiVersion: v1
clusters:
- cluster:
    server: https://premium.example.com:6443
    certificate-authority-data: Y2E=
  name: premium-cluster
contexts:
- context:
    cluster: premium-cluster
    user: replay
  name: replay
current-context: replay
users:
- name: replay
  user: {}`
)

var (
	_ replayClusterProvider = (*dedicatedReplayProvider)(nil)
	_ replayClusterProvider = (*premiumReplayProvider)(nil)
)

// mustNewReplayContext creates a replayContext for testing, failing the test if creation fails.
func mustNewReplayContext(t *testing.T, b *replayBuilder) *replayContext {
	t.Helper()
	rctx, err := newReplayContext(b.req, b.cfg, b.server)
	if err != nil {
		t.Fatalf("newReplayContext() error = %v", err)
	}
	return rctx
}

func replayErrorMatches(err, expected error) bool {
	if errors.Is(err, expected) {
		return true
	}
	if err == nil || expected == nil {
		return err == expected
	}
	return strings.Contains(err.Error(), expected.Error())
}

func TestReplayBuildErrorClassifiesDataplaneShapeErrorsAsInvalidArgument(t *testing.T) {
	tests := []struct {
		name string
		err  error
		want string
	}{
		{
			name: "missing auto dataplane data node",
			err:  errReplayDataplaneMissingDataNode,
			want: "data node is not set",
		},
		{
			name: "unsupported auto dataplane placement",
			err:  errReplayDataplaneUnsupportedPlacement,
			want: "not supported deploy position for filling dataplane info automatically",
		},
		{
			name: "unsupported dataplane cluster type",
			err:  unsupportedReplayClusterType(apiv1.Replay_CLUSTER_TYPE_UNSPECIFIED),
			want: "unsupported cluster type CLUSTER_TYPE_UNSPECIFIED for replay",
		},
		{
			name: "premium missing data node",
			err:  premiumReplayMissingDataNode(),
			want: "premium data node is not set",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			err := replayBuildError("Failed to build replay", tt.err)
			if status.Code(err) != codes.InvalidArgument {
				t.Fatalf("replayBuildError() code = %v, want %v", status.Code(err), codes.InvalidArgument)
			}
			if !strings.Contains(err.Error(), tt.want) {
				t.Fatalf("replayBuildError() error = %v, want substring %q", err, tt.want)
			}
		})
	}
}

func assertReplaySpecUnchanged(t *testing.T, got, want *cloudv1.Replay) {
	t.Helper()

	if !reflect.DeepEqual(got.Spec.SourceClusterType, want.Spec.SourceClusterType) ||
		!reflect.DeepEqual(got.Spec.SourceClusterID, want.Spec.SourceClusterID) ||
		!reflect.DeepEqual(got.Spec.SourceProvider, want.Spec.SourceProvider) ||
		!reflect.DeepEqual(got.Spec.SourceRegion, want.Spec.SourceRegion) ||
		got.Spec.TargetClusterType != want.Spec.TargetClusterType ||
		got.Spec.TargetClusterID != want.Spec.TargetClusterID ||
		got.Spec.TargetProvider != want.Spec.TargetProvider ||
		got.Spec.TargetRegion != want.Spec.TargetRegion ||
		!reflect.DeepEqual(got.Spec.Tasks, want.Spec.Tasks) ||
		!reflect.DeepEqual(got.Spec.ReplayConfig, want.Spec.ReplayConfig) ||
		!reflect.DeepEqual(got.Spec.Kubernetes, want.Spec.Kubernetes) {
		t.Fatalf("Replay execution spec mutated unexpectedly:\n got: %#v\nwant: %#v", got.Spec, want.Spec)
	}
}

func assertBuildForUpdateRejectedWithoutMutatingSpec(t *testing.T, builder *replayBuilder, replay *cloudv1.Replay) error {
	t.Helper()

	before := replay.DeepCopy()
	err := builder.BuildForUpdate(t.Context(), replay)
	if err == nil {
		t.Fatal("BuildForUpdate() error = nil, want rejected update")
	}
	assertReplaySpecUnchanged(t, replay, before)
	return err
}

func assertBuildForUpdateRejected(t *testing.T, builder *replayBuilder, replay *cloudv1.Replay) error {
	t.Helper()

	err := builder.BuildForUpdate(t.Context(), replay)
	if err == nil {
		t.Fatal("BuildForUpdate() error = nil, want rejected update")
	}
	return err
}

func assertUpdateReplayRejectedAndSpecUnchanged(
	t *testing.T,
	server *ReplayServer,
	kubeClient client.Client,
	namespace string,
	req *apiv1.Replay,
	original *cloudv1.Replay,
) error {
	t.Helper()

	_, err := server.UpdateReplay(t.Context(), &apiv1.UpdateReplayRequest{Replay: req})
	if err == nil {
		t.Fatal("UpdateReplay() error = nil, want rejected update")
	}

	updatedReplay := &cloudv1.Replay{}
	if getErr := kubeClient.Get(t.Context(), client.ObjectKey{Name: original.Name, Namespace: namespace}, updatedReplay); getErr != nil {
		t.Fatalf("Get(replay) error = %v", getErr)
	}
	assertReplaySpecUnchanged(t, updatedReplay, original)
	return err
}

func newPremiumBuildForUpdateRequest() *apiv1.Replay {
	return &apiv1.Replay{
		ReplayId:              "r1",
		ProjectId:             "p1",
		Provider:              consts.ProviderTypeACK,
		Region:                "ap-southeast-1",
		PremiumTargetEndpoint: ptr.To("tidb.request.example.com:4000"),
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aliyun{
					Aliyun: &apiv1.Replay_AliyunAuditLog{
						BucketUrl: "s3://request-audit/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
					},
				},
			},
		},
		SourceCluster: premiumReplayClusterRef(
			"source-premium",
			"p1",
			consts.ProviderTypeACK,
			"ap-southeast-1",
		),
		TargetCluster: premiumReplayClusterRef(
			"target-premium",
			"p1",
			consts.ProviderTypeACK,
			"ap-southeast-1",
		),
	}
}

func newPremiumBuildForUpdateReplay() *cloudv1.Replay {
	return &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			State:             cloudv1.ReplayStateDeploy,
			TargetClusterType: cloudv1.ReplayClusterTypePremium,
			SourceClusterID:   ptr.To("source-premium"),
			SourceClusterType: ptr.To(cloudv1.ReplayClusterTypePremium),
			SourceProvider:    ptr.To(consts.ProviderTypeACK),
			SourceRegion:      ptr.To("ap-southeast-1"),
			TargetClusterID:   "target-premium",
			TargetProvider:    consts.ProviderTypeACK,
			TargetRegion:      "ap-southeast-1",
		},
	}
}

func newPremiumUpdateReplayRequest() *apiv1.Replay {
	return &apiv1.Replay{
		ReplayId:              "replay-1",
		ProjectId:             "project-1",
		Provider:              consts.ProviderTypeACK,
		Region:                "ap-southeast-1",
		State:                 apiv1.Replay_STATE_DEPLOY,
		PremiumTargetEndpoint: ptr.To("tidb.request.example.com:4000"),
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aliyun{
					Aliyun: &apiv1.Replay_AliyunAuditLog{
						BucketUrl: "s3://request-audit/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
					},
				},
			},
		},
		SourceCluster: premiumReplayClusterRef(
			"source-premium",
			"project-1",
			consts.ProviderTypeACK,
			"ap-southeast-1",
		),
		TargetCluster: premiumReplayClusterRef(
			"target-premium",
			"project-1",
			consts.ProviderTypeACK,
			"ap-southeast-1",
		),
	}
}

func newPremiumUpdateReplay(namespace string) *cloudv1.Replay {
	replay := newPremiumBuildForUpdateReplay()
	replay.ObjectMeta = metav1.ObjectMeta{
		Name:      "replay-1",
		Namespace: namespace,
	}
	replay.Spec.SourceClusterID = ptr.To(replayOldSourceClusterID)
	return replay
}

func dedicatedReplayClusterRef(clusterID, projectID, provider, region string) *apiv1.Replay_ClusterRef {
	return &apiv1.Replay_ClusterRef{
		ClusterId:   clusterID,
		ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
		TenantId:    "tenant-1",
		ProjectId:   projectID,
		Provider:    provider,
		Region:      region,
	}
}

func premiumReplayClusterRef(clusterID, projectID, provider, region string) *apiv1.Replay_ClusterRef {
	return &apiv1.Replay_ClusterRef{
		ClusterId:   clusterID,
		ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
		TenantId:    "tenant-1",
		ProjectId:   projectID,
		Provider:    provider,
		Region:      region,
	}
}

func TestNewReplayClusterProviderDispatch(t *testing.T) {
	rctx := &replayContext{}

	for _, clusterType := range []apiv1.Replay_ClusterType{
		apiv1.Replay_CLUSTER_TYPE_DEDICATED,
		apiv1.Replay_CLUSTER_TYPE_PREMIUM,
	} {
		ref := &apiv1.Replay_ClusterRef{ClusterType: clusterType, ClusterId: "test"}
		p, err := newReplayClusterProvider(rctx, ref)
		if err != nil {
			t.Fatalf("newReplayClusterProvider(%v) error = %v", clusterType, err)
		}
		if p == nil {
			t.Fatalf("newReplayClusterProvider(%v) = nil, want provider", clusterType)
		}
		_ = p.LoadSourceMaterial
		_ = p.LoadTargetEndpoints
		_ = p.ResolveDataplane
	}

	_, err := newReplayClusterProvider(rctx, &apiv1.Replay_ClusterRef{ClusterType: apiv1.Replay_CLUSTER_TYPE_UNSPECIFIED, ClusterId: "x"})
	if err == nil {
		t.Fatal("newReplayClusterProvider(UNSPECIFIED) error = nil, want unsupported error")
	}
}

func TestNewReplayRouteStrategy(t *testing.T) {
	twoSourceUnits := []replaySourceTaskUnit{
		{name: "group-a", auditLogAddrs: []cloudv1.AuditLogAddr{{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://bucket/a"}}}},
		{name: "group-b", auditLogAddrs: []cloudv1.AuditLogAddr{{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://bucket/b"}}}},
	}
	tests := []struct {
		name       string
		sourceType apiv1.Replay_ClusterType
		targetType apiv1.Replay_ClusterType
		endpoints  []string
		wantErr    bool
		want       []replayTaskMaterial
	}{
		{
			name:       "dedicated to dedicated (1:1) preserves source unit names",
			sourceType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			targetType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			endpoints:  []string{"target-a:4000", "target-b:4000"},
			want: []replayTaskMaterial{
				{name: "group-a", targetEndpoint: "target-a:4000", auditLogAddrs: []cloudv1.AuditLogAddr{{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://bucket/a"}}}},
				{name: "group-b", targetEndpoint: "target-b:4000", auditLogAddrs: []cloudv1.AuditLogAddr{{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://bucket/b"}}}},
			},
		},
		{
			name:       "dedicated to premium (N:1) merges all audit logs",
			sourceType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			targetType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			endpoints:  []string{"premium.example.com:4000"},
			want: []replayTaskMaterial{
				{
					name:           replayTaskGroup0Name,
					targetEndpoint: "premium.example.com:4000",
					auditLogAddrs: []cloudv1.AuditLogAddr{
						{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://bucket/a"}},
						{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://bucket/b"}},
					},
				},
			},
		},
		{
			name:       "premium to premium (1:1)",
			sourceType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			targetType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			endpoints:  []string{"premium.example.com:4000"},
			want: []replayTaskMaterial{
				{
					name:           replayTaskGroup0Name,
					targetEndpoint: "premium.example.com:4000",
					auditLogAddrs: []cloudv1.AuditLogAddr{
						{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://bucket/a"}},
						{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://bucket/b"}},
					},
				},
			},
		},
		{
			name:       "premium to dedicated (N:1) merges all audit logs",
			sourceType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			targetType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			endpoints:  []string{"target.example.com:4000"},
			want: []replayTaskMaterial{
				{
					name:           replayTaskGroup0Name,
					targetEndpoint: "target.example.com:4000",
					auditLogAddrs: []cloudv1.AuditLogAddr{
						{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://bucket/a"}},
						{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://bucket/b"}},
					},
				},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			source := &replayClusterProviderStub{clusterType: tt.sourceType, loadSource: func(context.Context) (replaySourceMaterial, error) {
				return replaySourceMaterial{units: twoSourceUnits}, nil
			}}
			target := &replayClusterProviderStub{clusterType: tt.targetType, loadEndpoints: func(context.Context) ([]string, error) {
				return tt.endpoints, nil
			}}
			strategy, err := newReplayRouteStrategy(source, target)
			if tt.wantErr {
				if err == nil {
					t.Fatal("newReplayRouteStrategy() error = nil, want error")
				}
				return
			}
			if err != nil {
				t.Fatalf("newReplayRouteStrategy() error = %v", err)
			}
			materials, err := strategy.BuildTaskMaterials(t.Context())
			if err != nil {
				t.Fatalf("BuildTaskMaterials() error = %v", err)
			}
			if !reflect.DeepEqual(materials, tt.want) {
				t.Fatalf("BuildTaskMaterials() = %+v, want %+v", materials, tt.want)
			}
		})
	}
}

func TestRoundRobinReplayRouteStrategy(t *testing.T) {
	makeUnit := func(name, s3url string) replaySourceTaskUnit {
		return replaySourceTaskUnit{
			name:          name,
			auditLogAddrs: []cloudv1.AuditLogAddr{{AWS: &cloudv1.AWSAuditLog{S3URL: s3url}}},
		}
	}

	tests := []struct {
		name      string
		units     []replaySourceTaskUnit
		endpoints []string
		wantErr   bool
		want      []replayTaskMaterial
	}{
		{
			name:      "3->2 round-robin: A->x, B->y, C->x, merged audit logs",
			units:     []replaySourceTaskUnit{makeUnit("a", "s3://a"), makeUnit("b", "s3://b"), makeUnit("c", "s3://c")},
			endpoints: []string{"x:4000", "y:4000"},
			want: []replayTaskMaterial{
				{
					name:           "group-0",
					targetEndpoint: "x:4000",
					auditLogAddrs: []cloudv1.AuditLogAddr{
						{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://a"}},
						{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://c"}},
					},
				},
				{
					name:           "b",
					targetEndpoint: "y:4000",
					auditLogAddrs:  []cloudv1.AuditLogAddr{{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://b"}}},
				},
			},
		},
		{
			name:      "empty endpoint skipped: [x, '', z] + 3 sources -> round-robin over 2 valid endpoints",
			units:     []replaySourceTaskUnit{makeUnit("a", "s3://a"), makeUnit("b", "s3://b"), makeUnit("c", "s3://c")},
			endpoints: []string{"x:4000", "", "z:4000"},
			want: []replayTaskMaterial{
				{
					name:           "group-0",
					targetEndpoint: "x:4000",
					auditLogAddrs: []cloudv1.AuditLogAddr{
						{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://a"}},
						{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://c"}},
					},
				},
				{
					name:           "b",
					targetEndpoint: "z:4000",
					auditLogAddrs:  []cloudv1.AuditLogAddr{{AWS: &cloudv1.AWSAuditLog{S3URL: "s3://b"}}},
				},
			},
		},
		{
			name:      "no source units returns error",
			units:     nil,
			endpoints: []string{"x:4000"},
			wantErr:   true,
		},
		{
			name:      "all empty endpoints returns error",
			units:     []replaySourceTaskUnit{makeUnit("a", "s3://a")},
			endpoints: []string{"", ""},
			wantErr:   true,
		},
		{
			name:      "no endpoints returns error",
			units:     []replaySourceTaskUnit{makeUnit("a", "s3://a")},
			endpoints: nil,
			wantErr:   true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			source := &replayClusterProviderStub{loadSource: func(context.Context) (replaySourceMaterial, error) {
				return replaySourceMaterial{units: tt.units}, nil
			}}
			target := &replayClusterProviderStub{loadEndpoints: func(context.Context) ([]string, error) {
				return tt.endpoints, nil
			}}
			strategy := &roundRobinReplayRouteStrategy{source: source, target: target}
			materials, err := strategy.BuildTaskMaterials(t.Context())
			if tt.wantErr {
				if err == nil {
					t.Fatal("BuildTaskMaterials() error = nil, want error")
				}
				return
			}
			if err != nil {
				t.Fatalf("BuildTaskMaterials() error = %v", err)
			}
			if !reflect.DeepEqual(materials, tt.want) {
				t.Fatalf("BuildTaskMaterials() = %+v, want %+v", materials, tt.want)
			}
		})
	}
}

func TestBuildEndpoint(t *testing.T) {
	tests := []struct {
		name           string
		deployPosition string
		cluster        *cloudv1.Cluster
		groupID        string
		want           string
		wantErr        bool
	}{
		{
			name:           "OK - default group",
			deployPosition: DeployPositionTarget,
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						cloudv1.ProjectTagKey: "1",
					},
					Annotations: map[string]string{
						cloudv1.DefaultLoadBalancerIDAnnotationKey: "123",
					},
					Name: "123456789",
				},
			},
			groupID: "123",
			want:    "db-tidb.tidb123456789.svc:4000",
			wantErr: false,
		},
		{
			name:           "OK - extra group",
			deployPosition: DeployPositionTarget,
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						cloudv1.ProjectTagKey: "1",
					},
					Annotations: map[string]string{
						cloudv1.DefaultLoadBalancerIDAnnotationKey: "123",
					},
					Name: "123456789",
				},
				Status: cloudv1.ClusterStatus{
					ExtraEndpoints: []cloudv1.ExtraGroupStatus{
						{
							Name: "456",
							TiDB: &cloudv1.TiDBStatus{
								Members: []cloudv1.TiDBMember{
									{
										Name: "db-456-err8u8-tidb-1",
									},
								},
							},
						},
					},
				},
			},
			groupID: "456",
			want:    "db-456-err8u8-tidb.tidb123456789.svc:4000",
			wantErr: false,
		},
		{
			name:           "OK - private endpoint",
			deployPosition: DeployPositionSource,
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						cloudv1.ProjectTagKey: "1",
					},
					Annotations: map[string]string{
						cloudv1.DefaultLoadBalancerIDAnnotationKey: "123",
					},
					Name: "123456789",
				},
				Status: cloudv1.ClusterStatus{
					ExtraEndpoints: []cloudv1.ExtraGroupStatus{
						{
							Name: "456",
							SqlEndpointsStatus: cloudv1.SqlEndpointsStatus{
								PrivateEndpoint: &cloudv1.SqlEndpointStatus{
									Host: "private-tidb.jioweru89323.clusters.tidb-cloud.com",
									Port: 4000,
								},
							},
						},
					},
				},
			},
			groupID: "456",
			want:    "private-tidb.jioweru89323.clusters.tidb-cloud.com:4000",
			wantErr: false,
		},
		{
			name:           "Error - extra endpoint not found",
			deployPosition: DeployPositionTarget,
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						cloudv1.ProjectTagKey: "1",
					},
					Annotations: map[string]string{
						cloudv1.DefaultLoadBalancerIDAnnotationKey: "123",
					},
					Name: "123456789",
				},
				Status: cloudv1.ClusterStatus{
					ExtraEndpoints: []cloudv1.ExtraGroupStatus{
						{
							Name: "789",
							TiDB: &cloudv1.TiDBStatus{
								Members: []cloudv1.TiDBMember{
									{
										Name: "db-456-err8u8-tidb-1",
									},
								},
							},
						},
					},
				},
			},
			groupID: "456",
			wantErr: true,
		},
		{
			name:           "Error - extra endpoint not ready",
			deployPosition: DeployPositionTarget,
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						cloudv1.ProjectTagKey: "1",
					},
					Annotations: map[string]string{
						cloudv1.DefaultLoadBalancerIDAnnotationKey: "123",
					},
					Name: "123456789",
				},
				Status: cloudv1.ClusterStatus{
					ExtraEndpoints: []cloudv1.ExtraGroupStatus{
						{
							Name: "456",
							TiDB: &cloudv1.TiDBStatus{},
						},
					},
				},
			},
			groupID: "456",
			wantErr: true,
		},
		{
			name:           "Error - private endpoint not ready",
			deployPosition: DeployPositionSource,
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						cloudv1.ProjectTagKey: "1",
					},
					Annotations: map[string]string{
						cloudv1.DefaultLoadBalancerIDAnnotationKey: "123",
					},
					Name: "123456789",
				},
				Status: cloudv1.ClusterStatus{
					SqlEndpoints: cloudv1.SqlEndpointsStatus{},
				},
			},
			groupID: "123",
			wantErr: true,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			provider := &dedicatedReplayProvider{deployPosition: tt.deployPosition}
			got, gotErr := provider.buildEndpoint(tt.cluster, tt.groupID)
			if gotErr != nil {
				if !tt.wantErr {
					t.Errorf("buildEndpoint() failed: %v", gotErr)
				}
				return
			}
			if tt.wantErr {
				t.Fatal("buildEndpoint() succeeded unexpectedly")
			}
			if got != tt.want {
				t.Errorf("buildEndpoint() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestReplayBuilderDeployPosition(t *testing.T) {
	tests := []struct {
		name                string
		req                 *apiv1.Replay
		wantDeployPosition  string
		wantDataplaneRefID  string
		wantNilDataplaneRef bool
	}{
		{
			name: "dedicated target requires project match and falls through to source",
			req: &apiv1.Replay{
				ProjectId:     "project-1",
				TenantId:      "tenant-1",
				Provider:      consts.ProviderTypeACK,
				Region:        "ap-southeast-1",
				SourceCluster: dedicatedReplayClusterRef("source-1", "project-1", consts.ProviderTypeACK, "ap-southeast-1"),
				TargetCluster: dedicatedReplayClusterRef("target-1", "project-2", consts.ProviderTypeACK, "ap-southeast-1"),
			},
			wantDeployPosition: DeployPositionSource,
			wantDataplaneRefID: "source-1",
		},
		{
			name: "premium target also requires project match and falls through to source",
			req: &apiv1.Replay{
				ProjectId:     "project-1",
				TenantId:      "tenant-1",
				Provider:      consts.ProviderTypeACK,
				Region:        "ap-southeast-1",
				SourceCluster: dedicatedReplayClusterRef("source-1", "project-1", consts.ProviderTypeACK, "ap-southeast-1"),
				TargetCluster: premiumReplayClusterRef("target-premium", "project-2", consts.ProviderTypeACK, "ap-southeast-1"),
			},
			wantDeployPosition: DeployPositionSource,
			wantDataplaneRefID: "source-1",
		},
		{
			name: "no placement match defaults automatic dataplane to target",
			req: &apiv1.Replay{
				ProjectId:     "project-1",
				TenantId:      "tenant-1",
				Provider:      consts.ProviderTypeACK,
				Region:        "ap-southeast-1",
				SourceCluster: dedicatedReplayClusterRef("source-1", "project-2", consts.ProviderTypeACK, "ap-southeast-1"),
				TargetCluster: dedicatedReplayClusterRef("target-1", "project-2", consts.ProviderTypeACK, "ap-southeast-1"),
			},
			wantDeployPosition: DeployPositionTarget,
			wantDataplaneRefID: "target-1",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			deployPosition := computeReplayDeployPosition(tt.req)

			if deployPosition != tt.wantDeployPosition {
				t.Fatalf("computeReplayDeployPosition() = %q, want %q", deployPosition, tt.wantDeployPosition)
			}
			if tt.wantNilDataplaneRef {
				if deployPosition != DeployPositionIndividual {
					t.Fatalf("expected individual placement, got %q", deployPosition)
				}
				return
			}
			var dataplaneClusterRef *apiv1.Replay_ClusterRef
			switch deployPosition {
			case DeployPositionTarget:
				dataplaneClusterRef = tt.req.TargetCluster
			case DeployPositionSource:
				dataplaneClusterRef = tt.req.SourceCluster
			}
			if dataplaneClusterRef == nil || dataplaneClusterRef.ClusterId != tt.wantDataplaneRefID {
				t.Fatalf("dataplaneRef = %+v, want cluster id %q", dataplaneClusterRef, tt.wantDataplaneRefID)
			}
		})
	}
}

func TestReplayBuilderSetKubernetesInfoIndividualPlacementReturnsUnsupportedPlacement(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{ReplayId: "r1"})
	err := builder.setKubernetesInfo(t.Context(), &cloudv1.Replay{}, &replayContext{
		deployPosition: DeployPositionIndividual,
	})
	if !replayErrorMatches(err, errReplayDataplaneUnsupportedPlacement) {
		t.Fatalf("setKubernetesInfo() error = %v, want %v", err, errReplayDataplaneUnsupportedPlacement)
	}
}

func TestReplayBuilderSetKubernetesInfoRefreshesPremiumKubernetesWithoutPreservingExistingMetadata(t *testing.T) {
	origFn := newReplayClusterProviderFn
	t.Cleanup(func() { newReplayClusterProviderFn = origFn })
	newReplayClusterProviderFn = func(_ *replayContext, _ *apiv1.Replay_ClusterRef) (replayClusterProvider, error) {
		return &replayClusterProviderStub{
			resolveDataplane: func(context.Context) (replayDataplaneResolution, error) {
				return replayDataplaneResolution{
					kubernetes: cloudv1.Kubernetes{
						Name:     replayNewDataplaneName,
						Endpoint: "https://resolved.example.com:6443",
						Provider: consts.ProviderTypeACK,
						Account:  "123456",
					},
				}, nil
			},
		}, nil
	}

	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-1",
		TenantId:  "tenant-1",
		Provider:  consts.ProviderTypeACK,
		Region:    "ap-southeast-1",
		TargetCluster: premiumReplayClusterRef(
			"target-premium",
			"project-1",
			consts.ProviderTypeACK,
			"ap-southeast-1",
		),
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			Kubernetes: cloudv1.Kubernetes{
				Name:     "existing-dataplane",
				Provider: consts.ProviderTypeACK,
				Network:  "vpc-existing",
				Cloud: cloudv1.Cloud{
					Ack: &cloudv1.CloudAck{
						Subnets: []string{"vsw-existing"},
					},
				},
			},
		},
	}

	if err := builder.setKubernetesInfo(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("setKubernetesInfo() error = %v", err)
	}
	want := cloudv1.Kubernetes{
		Name:     replayNewDataplaneName,
		Endpoint: "https://resolved.example.com:6443",
		Provider: consts.ProviderTypeACK,
		Account:  "123456",
	}
	if !reflect.DeepEqual(replay.Spec.Kubernetes, want) {
		t.Fatalf("setKubernetesInfo() kubernetes = %#v, want %#v", replay.Spec.Kubernetes, want)
	}
}

func Test_getAuditlogAddr(t *testing.T) {
	tests := []struct {
		name string // description of this test case
		// Named input parameters for target function.
		replayReq *apiv1.Replay
		cluster   *cloudv1.Cluster
		groupID   string
		want      *cloudv1.AuditLogAddr
		wantErr   error
	}{
		{
			name: "OK - custom full auditlog",
			replayReq: &apiv1.Replay{
				Defaults: &apiv1.Replay_ReplayDefaults{
					SourceAuditLog: &apiv1.Replay_AuditLogAddr{
						Source: &apiv1.Replay_AuditLogAddr_Aws{
							Aws: &apiv1.Replay_AWSAuditLog{
								S3Url:      "s3://prod-xxxx-eb91457e/abc/123/",
								RoleArn:    "arn:aws:iam::2358495349:role/sb-audit-log-8fd12c83",
								ExternalId: ptr.To("test"),
							},
						},
					},
				},
			},
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						cloudv1.ProjectTagKey:     "1",
						cloudv1.ProviderTypeLabel: "eks",
					},
					Annotations: map[string]string{
						consts.AuditLogSidecarImageTag: "v1.0.0",
					},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						TiDB: cloudv1.TiDB{
							Audit: &cloudv1.Audit{
								Enable:    true,
								BucketURL: "s3://prod-xxxx-eb91457e/abc/456/",
								RoleArn:   "arn:aws:iam::2358495349:role/sb-audit-log-8fd12c83-readonly",
							},
						},
					},
				},
			},
			groupID: "123",
			want: &cloudv1.AuditLogAddr{
				AWS: &cloudv1.AWSAuditLog{
					S3URL:      "s3://prod-xxxx-eb91457e/abc/123/",
					RoleArn:    "arn:aws:iam::2358495349:role/sb-audit-log-8fd12c83",
					ExternalID: ptr.To("test"),
				},
			},
			wantErr: nil,
		},
		{
			name: "OK - custom auditlog addr",
			replayReq: &apiv1.Replay{
				Defaults: &apiv1.Replay_ReplayDefaults{
					SourceAuditLog: &apiv1.Replay_AuditLogAddr{
						Source: &apiv1.Replay_AuditLogAddr_Aws{
							Aws: &apiv1.Replay_AWSAuditLog{
								S3Url: "s3://prod-xxxx-eb91457e/abc/123/",
							},
						},
					},
				},
			},
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						cloudv1.ProjectTagKey:     "1",
						cloudv1.ProviderTypeLabel: "eks",
					},
					Annotations: map[string]string{
						consts.AuditLogSidecarImageTag: "v1.0.0",
					},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						TiDB: cloudv1.TiDB{
							Audit: &cloudv1.Audit{
								Enable:    true,
								BucketURL: "s3://prod-xxxx-eb91457e/abc/456/",
								RoleArn:   "arn:aws:iam::2358495349:role/sb-audit-log-8fd12c83-readonly",
							},
						},
					},
				},
			},
			groupID: "123",
			want: &cloudv1.AuditLogAddr{
				AWS: &cloudv1.AWSAuditLog{
					S3URL:      "s3://prod-xxxx-eb91457e/abc/123/",
					RoleArn:    "arn:aws:iam::2358495349:role/sb-audit-log-8fd12c83-readonly",
					ExternalID: ptr.To("1"),
				},
			},
			wantErr: nil,
		},
		{
			name: "OK - custom auditlog crendital",
			replayReq: &apiv1.Replay{
				Defaults: &apiv1.Replay_ReplayDefaults{
					SourceAuditLog: &apiv1.Replay_AuditLogAddr{
						Source: &apiv1.Replay_AuditLogAddr_Aws{
							Aws: &apiv1.Replay_AWSAuditLog{
								RoleArn:    "arn:aws:iam::2358495349:role/sb-audit-log-8fd12c83",
								ExternalId: ptr.To("test"),
							},
						},
					},
				},
			},
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						cloudv1.ProjectTagKey:     "1",
						cloudv1.ProviderTypeLabel: "eks",
					},
					Annotations: map[string]string{
						consts.AuditLogSidecarImageTag: "v1.0.0",
					},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						TiDB: cloudv1.TiDB{
							Audit: &cloudv1.Audit{
								Enable:    true,
								BucketURL: "s3://prod-xxxx-eb91457e/abc/456/",
								RoleArn:   "arn:aws:iam::2358495349:role/sb-audit-log-8fd12c83-readonly",
							},
						},
					},
				},
			},
			groupID: "123",
			want: &cloudv1.AuditLogAddr{
				AWS: &cloudv1.AWSAuditLog{
					S3URL:      "s3://prod-xxxx-eb91457e/abc/456/123_tidb-",
					RoleArn:    "arn:aws:iam::2358495349:role/sb-audit-log-8fd12c83",
					ExternalID: ptr.To("test"),
				},
			},
			wantErr: nil,
		},
		{
			name:      "OK - without custom",
			replayReq: &apiv1.Replay{},
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						cloudv1.ProjectTagKey:     "1",
						cloudv1.ProviderTypeLabel: "eks",
					},
					Annotations: map[string]string{
						consts.AuditLogSidecarImageTag: "v1.0.0",
					},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						TiDB: cloudv1.TiDB{
							Audit: &cloudv1.Audit{
								Enable:    true,
								BucketURL: "s3://prod-xxxx-eb91457e/abc/456/",
								RoleArn:   "arn:aws:iam::2358495349:role/sb-audit-log-8fd12c83-readonly",
							},
						},
					},
				},
			},
			groupID: "123",
			want: &cloudv1.AuditLogAddr{
				AWS: &cloudv1.AWSAuditLog{
					S3URL:      "s3://prod-xxxx-eb91457e/abc/456/123_tidb-",
					RoleArn:    "arn:aws:iam::2358495349:role/sb-audit-log-8fd12c83-readonly",
					ExternalID: ptr.To("1"),
				},
			},
			wantErr: nil,
		},
		{
			name:      "Error - dedicated ACK source auditlog is unsupported",
			replayReq: &apiv1.Replay{},
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						cloudv1.ProjectTagKey:     "1",
						cloudv1.ProviderTypeLabel: "ack",
					},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						TiDB: cloudv1.TiDB{
							Audit: &cloudv1.Audit{
								Enable:    true,
								BucketURL: "oss://prod-xxxx-eb91457e/abc/456/",
							},
						},
					},
				},
			},
			groupID: "123",
			wantErr: unsupportedReplayCloudProvider("ack"),
		},
		{
			name:      "Error - audit log disabled",
			replayReq: &apiv1.Replay{},
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name: "source-1",
					Labels: map[string]string{
						cloudv1.ProjectTagKey:     "1",
						cloudv1.ProviderTypeLabel: "eks",
					},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						TiDB: cloudv1.TiDB{
							Audit: &cloudv1.Audit{
								BucketURL: "s3://prod-xxxx-eb91457e/abc/456/",
							},
						},
					},
				},
			},
			groupID: "123",
			wantErr: errors.New("audit log is not enabled"),
		},
		{
			name:      "Error - unsupported provider",
			replayReq: &apiv1.Replay{},
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						cloudv1.ProjectTagKey:     "1",
						cloudv1.ProviderTypeLabel: "gke",
					},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						TiDB: cloudv1.TiDB{
							Audit: &cloudv1.Audit{
								Enable:    true,
								BucketURL: "s3://prod-xxxx-eb91457e/abc/456/",
								RoleArn:   "arn:aws:iam::2358495349:role/sb-audit-log-8fd12c83-readonly",
							},
						},
					},
				},
			},
			groupID: "123",
			wantErr: unsupportedReplayCloudProvider("gke"),
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if tt.replayReq.SourceCluster == nil {
				tt.replayReq.SourceCluster = &apiv1.Replay_ClusterRef{
					ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
					ClusterId:   "source-cluster",
					ProjectId:   "1",
				}
			}
			builder := newReplayBuilder(nil, tt.replayReq)
			var sourceAuditLog *apiv1.Replay_AuditLogAddr
			if builder.req.Defaults != nil {
				sourceAuditLog = builder.req.Defaults.SourceAuditLog
			}
			provider := &dedicatedReplayProvider{sourceAuditLog: sourceAuditLog}
			got, gotErr := provider.buildAuditLogAddr(tt.cluster, tt.groupID)
			if gotErr != nil {
				if tt.wantErr == nil {
					t.Errorf("getAuditlogAddr() failed: %v", gotErr)
					return
				}
				if !replayErrorMatches(gotErr, tt.wantErr) {
					t.Fatalf("getAuditlogAddr() error = %v, want %v", gotErr, tt.wantErr)
				}
				return
			}
			if tt.wantErr != nil {
				t.Fatalf("getAuditlogAddr() succeeded unexpectedly, want %v", tt.wantErr)
			}
			if !reflect.DeepEqual(got, *tt.want) {
				t.Errorf("getAuditlogAddr() = %+v, want %+v", got, *tt.want)
			}
		})
	}
}

type replayClusterProviderStub struct {
	clusterType      apiv1.Replay_ClusterType
	loadSource       func(context.Context) (replaySourceMaterial, error)
	loadEndpoints    func(context.Context) ([]string, error)
	resolveDataplane func(context.Context) (replayDataplaneResolution, error)
}

func (h *replayClusterProviderStub) GetClusterType() apiv1.Replay_ClusterType {
	return h.clusterType
}

func (h *replayClusterProviderStub) LoadSourceMaterial(ctx context.Context) (replaySourceMaterial, error) {
	if h.loadSource == nil {
		return replaySourceMaterial{}, nil
	}
	return h.loadSource(ctx)
}

func (h *replayClusterProviderStub) LoadTargetEndpoints(ctx context.Context) ([]string, error) {
	if h.loadEndpoints == nil {
		return nil, nil
	}
	return h.loadEndpoints(ctx)
}

func (h *replayClusterProviderStub) ResolveDataplane(ctx context.Context) (replayDataplaneResolution, error) {
	if h.resolveDataplane == nil {
		return replayDataplaneResolution{}, nil
	}
	return h.resolveDataplane(ctx)
}

func TestLoadPremiumSourceMaterial(t *testing.T) {
	provider := &premiumReplayProvider{
		sourceAuditLog: &apiv1.Replay_AuditLogAddr{
			Source: &apiv1.Replay_AuditLogAddr_Aws{
				Aws: &apiv1.Replay_AWSAuditLog{
					S3Url: "s3://request-audit/prefix",
				},
			},
		},
	}

	got, err := provider.LoadSourceMaterial(t.Context())
	if err != nil {
		t.Fatalf("LoadSourceMaterial() error = %v", err)
	}
	want := replaySourceMaterial{
		units: []replaySourceTaskUnit{
			{
				auditLogAddrs: []cloudv1.AuditLogAddr{
					{
						AWS: &cloudv1.AWSAuditLog{
							S3URL: "s3://request-audit/prefix",
						},
					},
				},
			},
		},
	}
	if !reflect.DeepEqual(got, want) {
		t.Fatalf("LoadSourceMaterial() = %#v, want %#v", got, want)
	}
}

func TestBuildExplicitDataplaneCloudSupportedProviders(t *testing.T) {
	tests := []struct {
		name      string
		provider  string
		request   *apiv1.Kubernetes
		wantCloud cloudv1.Cloud
	}{
		{
			name:     "eks",
			provider: consts.ProviderTypeEKS,
			request: &apiv1.Kubernetes{
				Info: &apiv1.Kubernetes_Aws_{
					Aws: &apiv1.Kubernetes_Aws{},
				},
			},
			wantCloud: cloudv1.Cloud{
				AWS: &cloudv1.CloudAWS{},
			},
		},
		{
			name:     "ack",
			provider: consts.ProviderTypeACK,
			request: &apiv1.Kubernetes{
				Info: &apiv1.Kubernetes_Alicloud_{
					Alicloud: &apiv1.Kubernetes_Alicloud{},
				},
			},
			wantCloud: cloudv1.Cloud{
				Ack: &cloudv1.CloudAck{},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := buildExplicitDataplaneCloud(tt.provider, tt.request)
			if err != nil {
				t.Fatalf("buildExplicitDataplaneCloud() error = %v", err)
			}
			if !reflect.DeepEqual(got, tt.wantCloud) {
				t.Fatalf("buildExplicitDataplaneCloud() = %#v, want %#v", got, tt.wantCloud)
			}
		})
	}
}

func buildPremiumReplayKubernetesForTest(
	t *testing.T,
	provider string,
) *cloudv1.Kubernetes {
	t.Helper()

	dataPlane := &dataplanemgrv1.DataPlane{
		AccountId: "123456",
		Kubernetes: &dataplanemgrv1.DataPlane_Kubernetes{
			Name:       replayNewDataplaneName,
			Version:    "1.29",
			Kubeconfig: replayPremiumTestKubeconfig,
		},
	}

	kubernetes, kubeconfigData, err := (&premiumReplayProvider{}).buildKubernetes(dataPlane, "dp-1", provider)
	if err != nil {
		t.Fatalf("buildKubernetes() error = %v", err)
	}
	if got := string(kubeconfigData); got != replayPremiumTestKubeconfig {
		t.Fatalf("kubeconfigData = %q, want premium replay kubeconfig", got)
	}
	if kubernetes.Provider != provider {
		t.Fatalf("Provider = %q, want %q", kubernetes.Provider, provider)
	}
	if kubernetes.Network != "" {
		t.Fatalf("Network = %q, want empty on full refresh", kubernetes.Network)
	}
	if kubernetes.Endpoint != "https://premium.example.com:6443" {
		t.Fatalf("Endpoint = %q, want kubeconfig endpoint", kubernetes.Endpoint)
	}
	if kubernetes.Name != "premium-cluster" {
		t.Fatalf("Name = %q, want kubeconfig cluster name", kubernetes.Name)
	}
	if kubernetes.Account != "123456" {
		t.Fatalf("Account = %q, want dataplane account id", kubernetes.Account)
	}
	return kubernetes
}

func TestBuildPremiumReplayKubernetesAppliesEKSProviderDefaults(t *testing.T) {
	kubernetes := buildPremiumReplayKubernetesForTest(t, consts.ProviderTypeEKS)

	aws := kubernetes.Cloud.AWS
	if aws == nil {
		t.Fatal("cloud.aws = nil, want EKS premium defaults")
	}

	groups := make(map[cloudv1.AWSSubnetGroupName]cloudv1.AWSSubnetGroup, len(aws.SubnetGroups))
	for _, group := range aws.SubnetGroups {
		groups[group.Name] = group
	}
	for _, name := range []cloudv1.AWSSubnetGroupName{
		cloudv1.AWSSubnetGroupNodes,
		cloudv1.AWSSubnetGroupPublicNLB,
		cloudv1.AWSSubnetGroupPrivateNLB,
	} {
		group, ok := groups[name]
		if !ok {
			t.Fatalf("missing AWS subnet group %q", name)
		}
		if group.Subnets == nil {
			t.Fatalf("AWS subnet group %q has nil subnets, want empty slice", name)
		}
	}
}

func TestBuildPremiumReplayKubernetesAppliesACKProviderDefaults(t *testing.T) {
	kubernetes := buildPremiumReplayKubernetesForTest(t, consts.ProviderTypeACK)
	if kubernetes.Cloud.Ack == nil {
		t.Fatal("cloud.ack = nil, want ACK premium defaults")
	}
}

func TestResolvePremiumReplayKubernetesRejectsMissingDataNodeOrUnsupportedProvider(t *testing.T) {
	tests := []struct {
		name     string
		dataNode *apiv1.Replay_ClusterRef
		wantErr  error
	}{
		{
			name:    "missing data node",
			wantErr: premiumReplayMissingDataNode(),
		},
		{
			name: "unsupported provider",
			dataNode: &apiv1.Replay_ClusterRef{
				ClusterId: "premium-1",
				Provider:  "gke",
			},
			wantErr: unsupportedReplayCloudProvider("gke"),
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			_, _, err := (&premiumReplayProvider{clusterRef: tt.dataNode}).resolveKubernetes(t.Context())
			if !replayErrorMatches(err, tt.wantErr) {
				t.Fatalf("resolveKubernetes() error = %v, want %v", err, tt.wantErr)
			}
		})
	}
}

func TestReplayBuilderApplyAuditCredentialOverrideRejectsMismatchedProviderType(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aws{
					Aws: &apiv1.Replay_AWSAuditLog{
						S3Url: "s3://bucket/prefix",
					},
				},
			},
		},
	})

	addr := &cloudv1.AuditLogAddr{
		Aliyun: &cloudv1.AliyunAuditLog{
			BucketURL: "s3://bucket/prefix?endpoint=http://oss-ap-southeast-1.aliyuncs.com",
		},
	}

	provider := &dedicatedReplayProvider{sourceAuditLog: mustNewReplayContext(t, builder).sourceAuditLog}
	err := provider.overrideSourceAuditLog(addr)
	if err == nil {
		t.Fatal("expected provider mismatch error, got nil")
	}
	if !replayErrorMatches(err, errReplaySourceAuditLogProviderMismatch) {
		t.Fatalf("expected provider mismatch sentinel, got %v", err)
	}
	if !strings.Contains(err.Error(), "provider mismatch") {
		t.Fatalf("expected mismatch error, got %v", err)
	}
}

func TestReplayBuilderBuildReplayTasksIgnoresNonAWSAuditLogOverrideForEKS(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	sourceNamespace := genNamespace(config2.GetConfig().NamespacePrefix, "source-project")
	targetNamespace := genNamespace(config2.GetConfig().NamespacePrefix, "target-project")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-1",
			Namespace: sourceNamespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "source-project",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://source-audit/prefix",
						RoleArn:   "arn:aws:iam::123456789012:role/replay",
					},
				},
			},
		},
	}
	// Target cluster must exist so LoadTargetEndpoints succeeds before source loading.
	targetCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "target-1",
			Namespace: targetNamespace,
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
				consts.AuditLogSidecarImageTag:             "v1.0.0",
			},
		},
		Status: cloudv1.ClusterStatus{
			SqlEndpoints: cloudv1.SqlEndpointsStatus{
				PrivateEndpoint: &cloudv1.SqlEndpointStatus{
					Host: "target.example.com",
					Port: 4000,
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster, targetCluster).
		Build()
	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "replay-project",
		Provider:  consts.ProviderTypeEKS,
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			ProjectId:   "source-project",
			Provider:    consts.ProviderTypeEKS,
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			ProjectId:   "target-project",
			Provider:    consts.ProviderTypeEKS,
		},
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aliyun{
					Aliyun: &apiv1.Replay_AliyunAuditLog{
						BucketUrl: "oss://source-audit/prefix",
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{ObjectMeta: metav1.ObjectMeta{Name: "r1"}}
	err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder))
	if err != nil {
		t.Fatalf("buildReplayTasks() error = %v", err)
	}
	if len(replay.Spec.Tasks) != 1 {
		t.Fatalf("buildReplayTasks() task count = %d, want 1", len(replay.Spec.Tasks))
	}
	addrs := replay.Spec.Tasks[0].SourceAuditLogAddrs
	if len(addrs) != 1 || addrs[0].AWS == nil {
		t.Fatalf("SourceAuditLogAddrs = %#v, want one AWS audit log", addrs)
	}
	if addrs[0].AWS.S3URL != "s3://source-audit/prefix/tidb-" {
		t.Fatalf("S3URL = %q, want cluster-derived AWS audit log", addrs[0].AWS.S3URL)
	}
}

func TestReplayBuilderBuildReplayTasksExplicitTasksShortCircuitRouteValidation(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		Provider:  consts.ProviderTypeEKS,
		Region:    "us-west-2",
		Tasks: []*apiv1.Replay_ReplayTask{
			{
				Name: replayTaskGroup0Name,
				Source: &apiv1.Replay_DataSource{
					AuditLogs: []*apiv1.Replay_AuditLogAddr{
						{
							Source: &apiv1.Replay_AuditLogAddr_Aws{
								Aws: &apiv1.Replay_AWSAuditLog{
									S3Url: "s3://manual-audit/prefix",
								},
							},
						},
					},
				},
				Target: &apiv1.Replay_DataTarget{
					Endpoint: "manual-target.example.com:4000",
				},
			},
		},
	})

	replay := &cloudv1.Replay{}
	if err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("buildReplayTasks() error = %v, want explicit tasks to short-circuit route validation", err)
	}
	if len(replay.Spec.Tasks) != 1 {
		t.Fatalf("task count = %d, want 1", len(replay.Spec.Tasks))
	}
	if replay.Spec.Tasks[0].Name != replayTaskGroup0Name {
		t.Fatalf("task name = %q, want %q", replay.Spec.Tasks[0].Name, replayTaskGroup0Name)
	}
	if replay.Spec.Tasks[0].TargetEndpoint != "manual-target.example.com:4000" {
		t.Fatalf("target endpoint = %q, want manual endpoint", replay.Spec.Tasks[0].TargetEndpoint)
	}
}

func TestReplayBuilderApplyAuditCredentialOverrideAWSRoleArnOnlyRejected(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aws{
					Aws: &apiv1.Replay_AWSAuditLog{
						RoleArn: replayNewAWSRoleARN,
					},
				},
			},
		},
	})

	addr := &cloudv1.AuditLogAddr{
		AWS: &cloudv1.AWSAuditLog{
			S3URL:      "s3://bucket/prefix/default_tidb-",
			RoleArn:    "arn:aws:iam::123456789012:role/replay-old",
			ExternalID: ptr.To("external-old"),
		},
	}

	provider := &dedicatedReplayProvider{sourceAuditLog: mustNewReplayContext(t, builder).sourceAuditLog}
	err := provider.overrideSourceAuditLog(addr)
	if !replayErrorMatches(err, errReplayAWSAuditLogCredentialMismatch) {
		t.Fatalf("overrideSourceAuditLog() error = %v, want credential mismatch", err)
	}
}

func TestReplayBuilderApplyAuditCredentialOverrideAWSExternalIDOnlyRejected(t *testing.T) {
	externalID := replayNewExternalID
	builder := newReplayBuilder(nil, &apiv1.Replay{
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aws{
					Aws: &apiv1.Replay_AWSAuditLog{
						ExternalId: &externalID,
					},
				},
			},
		},
	})

	addr := &cloudv1.AuditLogAddr{
		AWS: &cloudv1.AWSAuditLog{
			S3URL:      "s3://bucket/prefix/default_tidb-",
			RoleArn:    "arn:aws:iam::123456789012:role/replay-old",
			ExternalID: ptr.To("external-old"),
		},
	}

	provider := &dedicatedReplayProvider{sourceAuditLog: mustNewReplayContext(t, builder).sourceAuditLog}
	err := provider.overrideSourceAuditLog(addr)
	if !replayErrorMatches(err, errReplayAWSAuditLogCredentialMismatch) {
		t.Fatalf("overrideSourceAuditLog() error = %v, want credential mismatch", err)
	}
}

func TestValidateAWSAuditLogFieldCombinations(t *testing.T) {
	externalID := replayNewExternalID
	tests := []struct {
		name string
		aws  *apiv1.Replay_AWSAuditLog
		err  error
	}{
		{
			name: "s3 only",
			aws: &apiv1.Replay_AWSAuditLog{
				S3Url: "s3://source-audit/manual-prefix",
			},
		},
		{
			name: "role and external id only",
			aws: &apiv1.Replay_AWSAuditLog{
				RoleArn:    replayNewAWSRoleARN,
				ExternalId: &externalID,
			},
		},
		{
			name: "s3 with role and external id",
			aws: &apiv1.Replay_AWSAuditLog{
				S3Url:      "s3://source-audit/manual-prefix",
				RoleArn:    replayNewAWSRoleARN,
				ExternalId: &externalID,
			},
		},
		{
			name: "empty",
			aws:  &apiv1.Replay_AWSAuditLog{},
			err:  errReplayAWSAuditLogMissingSource,
		},
		{
			name: "role only",
			aws: &apiv1.Replay_AWSAuditLog{
				RoleArn: replayNewAWSRoleARN,
			},
			err: errReplayAWSAuditLogCredentialMismatch,
		},
		{
			name: "external id only",
			aws: &apiv1.Replay_AWSAuditLog{
				ExternalId: &externalID,
			},
			err: errReplayAWSAuditLogCredentialMismatch,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			err := validateAWSAuditLog(tt.aws)
			if !replayErrorMatches(err, tt.err) {
				t.Fatalf("validateAWSAuditLog() error = %v, want %v", err, tt.err)
			}
		})
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedSourcePremiumTargetFallsBackToSourceClusterAuditLog(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-1",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
				consts.AuditLogSidecarImageTag:             "v1.0.0",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://source-audit/new-prefix",
						RoleArn:   replayNewAWSRoleARN,
					},
				},
			},
		},
		Status: cloudv1.ClusterStatus{
			ExtraEndpoints: []cloudv1.ExtraGroupStatus{
				{
					Name:  "analytics",
					Phase: cloudv1.ClusterPhaseRunning,
					TiDB: &cloudv1.TiDBStatus{
						Members: []cloudv1.TiDBMember{{Name: "db-analytics-a-tidb-0"}},
					},
				},
				{
					Name:  "ad-hoc",
					Phase: cloudv1.ClusterPhaseRunning,
					TiDB: &cloudv1.TiDBStatus{
						Members: []cloudv1.TiDBMember{{Name: "db-ad-hoc-a-tidb-0"}},
					},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:              "r1",
		ProjectId:             "project-1",
		Provider:              consts.ProviderTypeACK,
		Region:                "ap-southeast-1",
		PremiumTargetEndpoint: ptr.To("tidb.new.example.com:4000"),
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-premium",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		Defaults: &apiv1.Replay_ReplayDefaults{
			Replicas: ptr.To(int32(3)),
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			SourceClusterID:   ptr.To("source-1"),
			SourceClusterType: ptr.To(cloudv1.ReplayClusterTypeDedicated),
			SourceProvider:    ptr.To(consts.ProviderTypeEKS),
			SourceRegion:      ptr.To("us-west-2"),
			TargetClusterType: cloudv1.ReplayClusterTypePremium,
			Tasks: []cloudv1.ReplayTask{
				{
					Name:           replayTaskGroup0Name,
					TargetEndpoint: "tidb.old.example.com:4000",
					Replicas:       1,
					SourceAuditLogAddrs: []cloudv1.AuditLogAddr{
						{
							AWS: &cloudv1.AWSAuditLog{
								S3URL:   replayOldPrefixDefaultS3URL,
								RoleArn: "arn:aws:iam::123456789012:role/replay-old",
							},
						},
						{
							AWS: &cloudv1.AWSAuditLog{
								S3URL:   "s3://source-audit/old-prefix/analytics_tidb-",
								RoleArn: "arn:aws:iam::123456789012:role/replay-old",
							},
						},
					},
				},
			},
		},
	}

	if err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("buildReplayTasks() error = %v", err)
	}
	if len(replay.Spec.Tasks) != 1 {
		t.Fatalf("len(tasks) = %d, want 1", len(replay.Spec.Tasks))
	}
	task := replay.Spec.Tasks[0]
	if task.Name != replayTaskGroup0Name {
		t.Fatalf("task.Name = %q, want group-0", task.Name)
	}
	if task.TargetEndpoint != "tidb.new.example.com:4000" {
		t.Fatalf("task.TargetEndpoint = %q, want premium target endpoint", task.TargetEndpoint)
	}
	if task.Replicas != 3 {
		t.Fatalf("task.Replicas = %d, want 3", task.Replicas)
	}
	if task.Resources != nil {
		t.Fatalf("task.Resources = %#v, want nil when request does not pass explicit resources", task.Resources)
	}
	wantS3URLs := []string{
		"s3://source-audit/new-prefix/default_tidb-",
		"s3://source-audit/new-prefix/analytics_tidb-",
		"s3://source-audit/new-prefix/ad-hoc_tidb-",
	}
	if len(task.SourceAuditLogAddrs) != len(wantS3URLs) {
		t.Fatalf("len(SourceAuditLogAddrs) = %d, want %d", len(task.SourceAuditLogAddrs), len(wantS3URLs))
	}
	for i, want := range wantS3URLs {
		if task.SourceAuditLogAddrs[i].AWS == nil || task.SourceAuditLogAddrs[i].AWS.S3URL != want {
			t.Fatalf("SourceAuditLogAddrs[%d] = %#v, want %s", i, task.SourceAuditLogAddrs[i], want)
		}
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedSourcePremiumTargetRequiresSourceProjectForClusterFallback(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	builder := newReplayBuilder(NewReplayServer(fake.NewClientBuilder().WithScheme(scheme.Scheme).Build()), &apiv1.Replay{
		ReplayId:              "r1",
		ProjectId:             "project-replay",
		Provider:              consts.ProviderTypeACK,
		Region:                "ap-southeast-1",
		PremiumTargetEndpoint: ptr.To("tidb.new.example.com:4000"),
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-premium",
			ProjectId:   "project-replay",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		Defaults: &apiv1.Replay_ReplayDefaults{
			Replicas: ptr.To(int32(2)),
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			SourceClusterID:   ptr.To("source-1"),
			SourceClusterType: ptr.To(cloudv1.ReplayClusterTypeDedicated),
			SourceProvider:    ptr.To(consts.ProviderTypeEKS),
			SourceRegion:      ptr.To("us-west-2"),
			TargetClusterType: cloudv1.ReplayClusterTypePremium,
			Tasks: []cloudv1.ReplayTask{
				{
					Name:           replayTaskGroup0Name,
					TargetEndpoint: "tidb.old.example.com:4000",
					Replicas:       1,
					SourceAuditLogAddrs: []cloudv1.AuditLogAddr{
						{
							AWS: &cloudv1.AWSAuditLog{
								S3URL:   replayOldPrefixDefaultS3URL,
								RoleArn: "arn:aws:iam::123456789012:role/replay-old",
							},
						},
					},
				},
			},
		},
	}

	err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder))
	if !replayErrorMatches(err, errReplaySourceClusterProjectMissing) {
		if err == nil || !strings.Contains(err.Error(), "source cluster is not set") {
			t.Fatalf("buildReplayTasks() error = %v, want source cluster load error", err)
		}
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedSourcePremiumTargetValidatesEndpointBeforeSourceLookup(t *testing.T) {
	builder := newReplayBuilder(NewReplayServer(nil), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-replay",
		Provider:  consts.ProviderTypeACK,
		Region:    "ap-southeast-1",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "missing-source",
			ProjectId:   "source-project",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-premium",
			ProjectId:   "project-replay",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
	})

	err := builder.buildReplayTasks(t.Context(), &cloudv1.Replay{}, mustNewReplayContext(t, builder))
	if !replayErrorMatches(err, errPremiumReplayMissingTargetEndpoint) {
		t.Fatalf("buildReplayTasks() error = %v, want %v", err, errPremiumReplayMissingTargetEndpoint)
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedSourcePremiumTargetAWSRoleOnlyUsesClusterAuditLog(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-1",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
				consts.AuditLogSidecarImageTag:             "v1.0.0",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://source-audit/new-prefix",
						RoleArn:   "arn:aws:iam::123456789012:role/replay-cluster",
					},
				},
			},
		},
		Status: cloudv1.ClusterStatus{
			ExtraEndpoints: []cloudv1.ExtraGroupStatus{
				{
					Name:  "analytics",
					Phase: cloudv1.ClusterPhaseRunning,
					TiDB: &cloudv1.TiDBStatus{
						Members: []cloudv1.TiDBMember{{Name: "db-analytics-a-tidb-0"}},
					},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster).
		Build()

	externalID := replayUpdatedExternalID
	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:              "r1",
		ProjectId:             "project-1",
		Provider:              consts.ProviderTypeACK,
		Region:                "ap-southeast-1",
		PremiumTargetEndpoint: ptr.To("tidb.new.example.com:4000"),
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-premium",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aws{
					Aws: &apiv1.Replay_AWSAuditLog{
						RoleArn:    replayNewAWSRoleARN,
						ExternalId: &externalID,
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{
			Name: "r1",
			Labels: map[string]string{
				cloudv1.ProjectTagKey: "project-1",
			},
		},
		Spec: cloudv1.ReplaySpec{
			SourceClusterID:   ptr.To("source-1"),
			SourceClusterType: ptr.To(cloudv1.ReplayClusterTypeDedicated),
			SourceProvider:    ptr.To(consts.ProviderTypeEKS),
			SourceRegion:      ptr.To("us-west-2"),
			TargetClusterType: cloudv1.ReplayClusterTypePremium,
			Tasks: []cloudv1.ReplayTask{
				{
					Name:           replayTaskGroup0Name,
					TargetEndpoint: "tidb.old.example.com:4000",
					Replicas:       1,
					SourceAuditLogAddrs: []cloudv1.AuditLogAddr{
						{
							AWS: &cloudv1.AWSAuditLog{
								S3URL:      replayOldPrefixDefaultS3URL,
								RoleArn:    "arn:aws:iam::123456789012:role/replay-old",
								ExternalID: ptr.To("old-external-id"),
							},
						},
						{
							AWS: &cloudv1.AWSAuditLog{
								S3URL:      "s3://source-audit/old-prefix/analytics_tidb-",
								RoleArn:    "arn:aws:iam::123456789012:role/replay-old",
								ExternalID: ptr.To("old-external-id"),
							},
						},
						{
							AWS: &cloudv1.AWSAuditLog{
								S3URL:      "s3://source-audit/old-prefix/ad-hoc_tidb-",
								RoleArn:    "arn:aws:iam::123456789012:role/replay-old",
								ExternalID: ptr.To("old-external-id"),
							},
						},
					},
				},
			},
		},
	}

	if err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("buildReplayTasks() error = %v", err)
	}
	if len(replay.Spec.Tasks) != 1 {
		t.Fatalf("len(tasks) = %d, want 1", len(replay.Spec.Tasks))
	}
	addrs := replay.Spec.Tasks[0].SourceAuditLogAddrs
	wantS3URLs := []string{
		"s3://source-audit/new-prefix/default_tidb-",
		"s3://source-audit/new-prefix/analytics_tidb-",
	}
	if len(addrs) != len(wantS3URLs) {
		t.Fatalf("len(SourceAuditLogAddrs) = %d, want %d", len(addrs), len(wantS3URLs))
	}
	for i, want := range wantS3URLs {
		if addrs[i].AWS == nil || addrs[i].AWS.S3URL != want {
			t.Fatalf("SourceAuditLogAddrs[%d] = %#v, want %s", i, addrs[i], want)
		}
		if addrs[i].AWS.RoleArn != replayNewAWSRoleARN {
			t.Fatalf("SourceAuditLogAddrs[%d].AWS.RoleArn = %q, want %q", i, addrs[i].AWS.RoleArn, replayNewAWSRoleARN)
		}
		if addrs[i].AWS.ExternalID == nil || *addrs[i].AWS.ExternalID != replayUpdatedExternalID {
			t.Fatalf("SourceAuditLogAddrs[%d].AWS.ExternalID = %v, want %q", i, addrs[i].AWS.ExternalID, replayUpdatedExternalID)
		}
	}
}

func TestReplayBuilderBuildReplayTasksCreatePathStillValidatesPremiumInputs(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    "ack",
			Region:      "ap-southeast-1",
		},
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aliyun{
					Aliyun: &apiv1.Replay_AliyunAuditLog{
						BucketUrl: "s3://bucket/prefix",
					},
				},
			},
		},
		PremiumTargetEndpoint: ptr.To("tidb.example.com:4000"),
	})

	replay := &cloudv1.Replay{}

	err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder))
	if err == nil {
		t.Fatalf("buildReplayTasks() error = nil, want source cluster validation error")
	}
	if !strings.Contains(err.Error(), "source cluster is not set") {
		t.Fatalf("buildReplayTasks() error = %v, want source cluster validation error", err)
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedSourcePremiumTargetFallsBackToClusterAuditLogWithoutOverride(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-1",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
				consts.AuditLogSidecarImageTag:             "v1.0.0",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://source-audit/prefix",
						RoleArn:   "arn:aws:iam::123456789012:role/replay",
					},
				},
			},
		},
		Status: cloudv1.ClusterStatus{
			ExtraEndpoints: []cloudv1.ExtraGroupStatus{
				{
					Name:  "analytics",
					Phase: cloudv1.ClusterPhaseRunning,
					TiDB: &cloudv1.TiDBStatus{
						Members: []cloudv1.TiDBMember{{Name: "db-analytics-a-tidb-0"}},
					},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-1",
		Provider:  consts.ProviderTypeACK,
		Region:    "ap-southeast-1",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-premium",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		PremiumTargetEndpoint: ptr.To("tidb.example.com:4000"),
		Defaults: &apiv1.Replay_ReplayDefaults{
			Replicas: ptr.To(int32(2)),
			Credential: &apiv1.Replay_DatabaseCredential{
				Username: "root",
			},
		},
	})

	replay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{Name: "r1"},
	}
	if err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("buildReplayTasks() error = %v", err)
	}
	if len(replay.Spec.Tasks) != 1 {
		t.Fatalf("len(tasks) = %d, want 1", len(replay.Spec.Tasks))
	}
	task := replay.Spec.Tasks[0]
	if task.Username != "root" {
		t.Fatalf("task.Username = %q, want root", task.Username)
	}
	wantS3URLs := []string{
		"s3://source-audit/prefix/default_tidb-",
		"s3://source-audit/prefix/analytics_tidb-",
	}
	if len(task.SourceAuditLogAddrs) != len(wantS3URLs) {
		t.Fatalf("len(SourceAuditLogAddrs) = %d, want %d", len(task.SourceAuditLogAddrs), len(wantS3URLs))
	}
	for i, want := range wantS3URLs {
		if task.SourceAuditLogAddrs[i].AWS == nil || task.SourceAuditLogAddrs[i].AWS.S3URL != want {
			t.Fatalf("SourceAuditLogAddrs[%d] = %#v, want %s", i, task.SourceAuditLogAddrs[i], want)
		}
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedSourcePremiumTargetExplicitAWSS3URLUsesSingleInput(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-1",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://cluster-audit/prefix",
						RoleArn:   "arn:aws:iam::123456789012:role/replay",
					},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:              "r1",
		ProjectId:             "project-1",
		Provider:              consts.ProviderTypeACK,
		Region:                "ap-southeast-1",
		PremiumTargetEndpoint: ptr.To("tidb.example.com:4000"),
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-premium",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aws{
					Aws: &apiv1.Replay_AWSAuditLog{
						S3Url:      "s3://source-audit/manual-prefix",
						RoleArn:    replayNewAWSRoleARN,
						ExternalId: ptr.To(replayNewExternalID),
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{ObjectMeta: metav1.ObjectMeta{Name: "r1"}}
	if err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("buildReplayTasks() error = %v", err)
	}

	if len(replay.Spec.Tasks) != 1 {
		t.Fatalf("len(tasks) = %d, want 1", len(replay.Spec.Tasks))
	}
	task := replay.Spec.Tasks[0]
	if len(task.SourceAuditLogAddrs) != 1 {
		t.Fatalf("len(SourceAuditLogAddrs) = %d, want 1", len(task.SourceAuditLogAddrs))
	}
	if task.SourceAuditLogAddrs[0].AWS == nil || task.SourceAuditLogAddrs[0].AWS.S3URL != "s3://source-audit/manual-prefix" {
		t.Fatalf("SourceAuditLogAddrs[0] = %#v, want single explicit aws audit log", task.SourceAuditLogAddrs[0])
	}
	if task.SourceAuditLogAddrs[0].AWS.RoleArn != replayNewAWSRoleARN {
		t.Fatalf("SourceAuditLogAddrs[0].AWS.RoleArn = %q, want override role arn", task.SourceAuditLogAddrs[0].AWS.RoleArn)
	}
	if task.SourceAuditLogAddrs[0].AWS.ExternalID == nil || *task.SourceAuditLogAddrs[0].AWS.ExternalID != replayNewExternalID {
		t.Fatalf("SourceAuditLogAddrs[0].AWS.ExternalID = %v, want %q", task.SourceAuditLogAddrs[0].AWS.ExternalID, replayNewExternalID)
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedSourcePremiumTargetExplicitAWSS3URLRequiresSourceProject(t *testing.T) {
	builder := newReplayBuilder(NewReplayServer(fake.NewClientBuilder().WithScheme(scheme.Scheme).Build()), &apiv1.Replay{
		ReplayId:              "r1",
		ProjectId:             "project-1",
		Provider:              consts.ProviderTypeACK,
		Region:                "ap-southeast-1",
		PremiumTargetEndpoint: ptr.To("tidb.example.com:4000"),
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-premium",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aws{
					Aws: &apiv1.Replay_AWSAuditLog{
						S3Url: "s3://source-audit/manual-prefix",
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{ObjectMeta: metav1.ObjectMeta{Name: "r1"}}
	err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder))
	if !replayErrorMatches(err, errReplaySourceClusterProjectMissing) {
		if err == nil || !strings.Contains(err.Error(), "source cluster is not set") {
			t.Fatalf("buildReplayTasks() error = %v, want source cluster load error", err)
		}
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedSourceDedicatedTargetExplicitAWSS3URLUsesSourceClusterShape(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-1",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
				consts.AuditLogSidecarImageTag:             "v1.0.0",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://cluster-audit/prefix",
						RoleArn:   "arn:aws:iam::123456789012:role/replay",
					},
				},
			},
		},
		Status: cloudv1.ClusterStatus{
			ExtraEndpoints: []cloudv1.ExtraGroupStatus{
				{
					Name:  "analytics",
					Phase: cloudv1.ClusterPhaseRunning,
					TiDB: &cloudv1.TiDBStatus{
						Members: []cloudv1.TiDBMember{
							{Name: "db-analytics-tidb-0"},
						},
					},
				},
			},
		},
	}
	targetCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "target-1",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
		Status: cloudv1.ClusterStatus{
			ExtraEndpoints: []cloudv1.ExtraGroupStatus{
				{
					Name:  "analytics",
					Phase: cloudv1.ClusterPhaseRunning,
					TiDB: &cloudv1.TiDBStatus{
						Members: []cloudv1.TiDBMember{
							{Name: "db-analytics-target-tidb-0"},
						},
					},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster, targetCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-1",
		Provider:  consts.ProviderTypeEKS,
		Region:    "us-west-2",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aws{
					Aws: &apiv1.Replay_AWSAuditLog{
						S3Url: "s3://source-audit/manual-prefix",
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{ObjectMeta: metav1.ObjectMeta{Name: "r1"}}
	if err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("buildReplayTasks() error = %v", err)
	}

	if len(replay.Spec.Tasks) != 1 {
		t.Fatalf("len(tasks) = %d, want 1", len(replay.Spec.Tasks))
	}
	task := replay.Spec.Tasks[0]
	if task.Name != replayTaskGroupDefaultName {
		t.Fatalf("Name = %q, want dedicated source group name", task.Name)
	}
	if task.TargetEndpoint != "db-tidb.tidbtarget-1.svc:4000" {
		t.Fatalf("TargetEndpoint = %q, want dedicated target internal endpoint", task.TargetEndpoint)
	}
	if len(task.SourceAuditLogAddrs) != 1 {
		t.Fatalf("len(SourceAuditLogAddrs) = %d, want 1", len(task.SourceAuditLogAddrs))
	}
	if task.SourceAuditLogAddrs[0].AWS == nil || task.SourceAuditLogAddrs[0].AWS.S3URL != "s3://source-audit/manual-prefix" {
		t.Fatalf("SourceAuditLogAddrs[0] = %#v, want single explicit aws audit log", task.SourceAuditLogAddrs[0])
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedSourcePremiumTargetAWSRoleAndExternalIDOnlyFallsBackToClusterAuditLog(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-1",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
				consts.AuditLogSidecarImageTag:             "v1.0.0",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://source-audit/prefix",
						RoleArn:   "arn:aws:iam::123456789012:role/replay-old",
					},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster).
		Build()

	externalID := replayNewExternalID
	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-1",
		Provider:  consts.ProviderTypeACK,
		Region:    "ap-southeast-1",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-premium",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		PremiumTargetEndpoint: ptr.To("tidb.example.com:4000"),
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aws{
					Aws: &apiv1.Replay_AWSAuditLog{
						RoleArn:    replayNewAWSRoleARN,
						ExternalId: &externalID,
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{Name: "r1"},
	}
	if err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("buildReplayTasks() error = %v", err)
	}
	addrs := replay.Spec.Tasks[0].SourceAuditLogAddrs
	if len(addrs) != 1 || addrs[0].AWS == nil {
		t.Fatalf("SourceAuditLogAddrs = %#v, want one AWS audit log", addrs)
	}
	if addrs[0].AWS.S3URL != "s3://source-audit/prefix/default_tidb-" {
		t.Fatalf("S3URL = %q, want cluster-derived prefix", addrs[0].AWS.S3URL)
	}
	if addrs[0].AWS.RoleArn != replayNewAWSRoleARN {
		t.Fatalf("RoleArn = %q, want %q", addrs[0].AWS.RoleArn, replayNewAWSRoleARN)
	}
	if addrs[0].AWS.ExternalID == nil || *addrs[0].AWS.ExternalID != replayNewExternalID {
		t.Fatalf("ExternalID = %v, want %q", addrs[0].AWS.ExternalID, replayNewExternalID)
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedSourcePremiumTargetAWSExternalIDOnlyRejected(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-1",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://source-audit/prefix",
						RoleArn:   "arn:aws:iam::123456789012:role/replay-old",
					},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster).
		Build()

	overrideExternalID := replayNewExternalID
	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-1",
		Provider:  consts.ProviderTypeACK,
		Region:    "ap-southeast-1",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-premium",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		PremiumTargetEndpoint: ptr.To("tidb.example.com:4000"),
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aws{
					Aws: &apiv1.Replay_AWSAuditLog{
						ExternalId: &overrideExternalID,
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{Name: "r1"},
	}
	err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder))
	if !replayErrorMatches(err, errReplayAWSAuditLogCredentialMismatch) {
		t.Fatalf("buildReplayTasks() error = %v, want credential mismatch", err)
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedSourcePremiumTargetRejectsDisabledClusterAudit(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-1",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-1",
		Provider:  consts.ProviderTypeACK,
		Region:    "ap-southeast-1",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-premium",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		PremiumTargetEndpoint: ptr.To("tidb.example.com:4000"),
	})

	replay := &cloudv1.Replay{ObjectMeta: metav1.ObjectMeta{Name: "r1"}}
	err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder))
	if err == nil || !strings.Contains(err.Error(), "audit log is not enabled") {
		t.Fatalf("buildReplayTasks() error = %v, want disabled audit log error", err)
	}
}

func TestReplayBuilderBuildReplayTasksACKDedicatedSourcePremiumTargetReturnsUnsupportedProvider(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-ack-dedicated",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeACK,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://source-audit/prefix?endpoint=http://oss-ap-southeast-1.aliyuncs.com/",
					},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-1",
		Provider:  consts.ProviderTypeACK,
		Region:    "ap-southeast-1",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-ack-dedicated",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-premium",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		PremiumTargetEndpoint: ptr.To("tidb.example.com:4000"),
		Defaults: &apiv1.Replay_ReplayDefaults{
			Replicas: ptr.To(int32(1)),
			Credential: &apiv1.Replay_DatabaseCredential{
				Username: "root",
			},
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aliyun{
					Aliyun: &apiv1.Replay_AliyunAuditLog{
						BucketUrl: "s3://source-audit/prefix?endpoint=http://oss-ap-southeast-1.aliyuncs.com/",
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{ObjectMeta: metav1.ObjectMeta{Name: "r1"}}
	err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder))
	if err == nil || !strings.Contains(err.Error(), "unsupported cloud provider ack for replay") {
		t.Fatalf("buildReplayTasks() error = %v, want unsupported ACK provider error", err)
	}
}

func TestReplayBuilderBuildReplayTasksACKDedicatedSourcePremiumTargetUnifiedAuditLogsReturnsUnsupportedProvider(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-ack-dedicated",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeACK,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "oss://source-audit/prefix/",
					},
				},
			},
		},
		Status: cloudv1.ClusterStatus{
			ExtraEndpoints: []cloudv1.ExtraGroupStatus{
				{
					Name:  "analytics",
					Phase: cloudv1.ClusterPhaseRunning,
					TiDB: &cloudv1.TiDBStatus{
						Members: []cloudv1.TiDBMember{{Name: "db-analytics-a-tidb-0"}},
					},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-1",
		Provider:  consts.ProviderTypeACK,
		Region:    "ap-southeast-1",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-ack-dedicated",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-premium",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		PremiumTargetEndpoint: ptr.To("tidb.example.com:4000"),
		Defaults: &apiv1.Replay_ReplayDefaults{
			Replicas: ptr.To(int32(1)),
			Credential: &apiv1.Replay_DatabaseCredential{
				Username: "root",
			},
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aliyun{
					Aliyun: &apiv1.Replay_AliyunAuditLog{
						BucketUrl: "s3://source-audit/prefix/tidb-?endpoint=http://oss-ap-southeast-1.aliyuncs.com&access-key=test-ak&secret-access-key=test-sk",
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{ObjectMeta: metav1.ObjectMeta{Name: "r1"}}
	err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder))
	if err == nil || !strings.Contains(err.Error(), "unsupported cloud provider ack for replay") {
		t.Fatalf("buildReplayTasks() error = %v, want unsupported ACK provider error", err)
	}
}

func TestReplayBuilderBuildReplayTasksACKDedicatedSourceDedicatedTargetReturnsUnsupportedProvider(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-ack-dedicated",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeACK,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "oss://source-audit/prefix/",
					},
				},
			},
		},
	}
	targetCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "target-ack-dedicated",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeACK,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster, targetCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		TenantId:  "tenant-1",
		ProjectId: "project-1",
		Provider:  consts.ProviderTypeACK,
		Region:    "ap-southeast-1",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-ack-dedicated",
			TenantId:    "tenant-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-ack-dedicated",
			TenantId:    "tenant-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
	})

	replay := &cloudv1.Replay{ObjectMeta: metav1.ObjectMeta{Name: "r1"}}
	err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder))
	if err == nil || !strings.Contains(err.Error(), "unsupported cloud provider ack for replay") {
		t.Fatalf("buildReplayTasks() error = %v, want unsupported ACK provider error", err)
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedToDedicatedBuildsMultipleGroups(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-1",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
				consts.AuditLogSidecarImageTag:             "v1.0.0",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://source-audit/prefix",
						RoleArn:   "arn:aws:iam::123456789012:role/replay",
					},
				},
			},
		},
		Status: cloudv1.ClusterStatus{
			ExtraEndpoints: []cloudv1.ExtraGroupStatus{
				{
					Name:  "analytics",
					Phase: cloudv1.ClusterPhaseRunning,
					TiDB: &cloudv1.TiDBStatus{
						Members: []cloudv1.TiDBMember{{Name: "db-analytics-source-tidb-0"}},
					},
				},
			},
		},
	}
	targetCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "target-1",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
		Status: cloudv1.ClusterStatus{
			ExtraEndpoints: []cloudv1.ExtraGroupStatus{
				{
					Name:  "analytics",
					Phase: cloudv1.ClusterPhaseRunning,
					TiDB: &cloudv1.TiDBStatus{
						Members: []cloudv1.TiDBMember{{Name: "db-analytics-target-tidb-0"}},
					},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster, targetCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		TenantId:  "tenant-1",
		ProjectId: "project-1",
		Provider:  consts.ProviderTypeEKS,
		Region:    "us-west-2",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			TenantId:    "tenant-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-1",
			TenantId:    "tenant-1",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
	})

	replay := &cloudv1.Replay{ObjectMeta: metav1.ObjectMeta{Name: "r1"}}
	if err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("buildReplayTasks() error = %v", err)
	}
	if len(replay.Spec.Tasks) != 2 {
		t.Fatalf("len(tasks) = %d, want 2", len(replay.Spec.Tasks))
	}

	tests := []struct {
		idx            int
		name           string
		targetEndpoint string
		auditLogS3URL  string
	}{
		{
			idx:            0,
			name:           replayTaskGroupDefaultName,
			targetEndpoint: "db-tidb.tidbtarget-1.svc:4000",
			auditLogS3URL:  replaySourceDefaultS3URL,
		},
		{
			idx:            1,
			name:           "group-analytics",
			targetEndpoint: "db-analytics-target-tidb.tidbtarget-1.svc:4000",
			auditLogS3URL:  "s3://source-audit/prefix/analytics_tidb-",
		},
	}
	for _, tt := range tests {
		task := replay.Spec.Tasks[tt.idx]
		if task.Name != tt.name {
			t.Fatalf("tasks[%d].Name = %q, want %q", tt.idx, task.Name, tt.name)
		}
		if task.TargetEndpoint != tt.targetEndpoint {
			t.Fatalf("tasks[%d].TargetEndpoint = %q, want %q", tt.idx, task.TargetEndpoint, tt.targetEndpoint)
		}
		if len(task.SourceAuditLogAddrs) != 1 || task.SourceAuditLogAddrs[0].AWS == nil {
			t.Fatalf("tasks[%d].SourceAuditLogAddrs = %#v, want one AWS audit log", tt.idx, task.SourceAuditLogAddrs)
		}
		if task.SourceAuditLogAddrs[0].AWS.S3URL != tt.auditLogS3URL {
			t.Fatalf("tasks[%d].SourceAuditLogAddrs[0].AWS.S3URL = %q, want %q",
				tt.idx,
				task.SourceAuditLogAddrs[0].AWS.S3URL,
				tt.auditLogS3URL,
			)
		}
	}
}

func TestReplayRouteCombinationRejectsUnsupportedClusterType(t *testing.T) {
	err := validateReplayRouteCombination(
		apiv1.Replay_CLUSTER_TYPE_UNSPECIFIED,
		apiv1.Replay_CLUSTER_TYPE_DEDICATED,
	)
	if err == nil {
		t.Fatal("buildReplayTasks() error = nil, want unsupported cluster type error")
	}
	if !strings.Contains(err.Error(), "unsupported route strategy") {
		t.Fatalf("buildReplayTasks() error = %v, want unsupported cluster type error", err)
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedToDedicatedWithExplicitAuditStillRequiresSourceCluster(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	targetCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "target-ack",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeACK,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(targetCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-1",
		Provider:  consts.ProviderTypeACK,
		Region:    "ap-southeast-1",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-eks",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-ack",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aws{
					Aws: &apiv1.Replay_AWSAuditLog{
						S3Url: "s3://source-audit/default_tidb-",
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{}
	err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder))
	if err == nil {
		t.Fatal("buildReplayTasks() error = nil, want source cluster lookup error")
	}
	if !strings.Contains(err.Error(), "source cluster is not set") {
		t.Fatalf("buildReplayTasks() error = %v, want source cluster lookup error", err)
	}
}

func TestReplayBuilderBuildReplayTasksACKDedicatedTargetMatchesMainBuilderBehavior(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-eks",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
				consts.AuditLogSidecarImageTag:             "v1.0.0",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://cluster-audit/prefix",
						RoleArn:   "arn:aws:iam::123456789012:role/replay",
					},
				},
			},
		},
	}
	targetCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "target-ack",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeACK,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster, targetCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-1",
		Provider:  consts.ProviderTypeACK,
		Region:    "ap-southeast-1",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-eks",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-ack",
			ProjectId:   "project-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aws{
					Aws: &apiv1.Replay_AWSAuditLog{
						S3Url: "s3://source-audit/default_tidb-",
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{}
	if err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("buildReplayTasks() error = %v", err)
	}
	if len(replay.Spec.Tasks) != 1 {
		t.Fatalf("tasks len = %d, want 1", len(replay.Spec.Tasks))
	}
	if got := replay.Spec.Tasks[0].Name; got != replayTaskGroupDefaultName {
		t.Fatalf("task name = %q, want group-default", got)
	}
	if got := replay.Spec.Tasks[0].TargetEndpoint; got != "db-tidb.tidbtarget-ack.svc:4000" {
		t.Fatalf("TargetEndpoint = %q, want ACK dedicated internal endpoint", got)
	}
}

func TestReplayBuilderBuildReplayTasksDedicatedToDedicatedSkipsInvalidExtraTargetGroup(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	sourceNamespace := genNamespace(config2.GetConfig().NamespacePrefix, "source-project")
	targetNamespace := genNamespace(config2.GetConfig().NamespacePrefix, "target-project")

	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-1",
			Namespace: sourceNamespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "source-project",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
				consts.AuditLogSidecarImageTag:             "v1.0.0",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://source-audit/prefix",
						RoleArn:   "arn:aws:iam::123456789012:role/replay",
					},
				},
			},
		},
		Status: cloudv1.ClusterStatus{
			ExtraEndpoints: []cloudv1.ExtraGroupStatus{
				{
					Name:  "analytics",
					Phase: cloudv1.ClusterPhaseRunning,
					TiDB: &cloudv1.TiDBStatus{
						Members: []cloudv1.TiDBMember{
							{Name: "db-analytics-tidb-0"},
						},
					},
				},
			},
		},
	}
	targetCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "target-1",
			Namespace: targetNamespace,
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
		Status: cloudv1.ClusterStatus{
			SqlEndpoints: cloudv1.SqlEndpointsStatus{
				PrivateEndpoint: &cloudv1.SqlEndpointStatus{
					Host: "target-default.example.com",
					Port: 4000,
				},
			},
			ExtraEndpoints: []cloudv1.ExtraGroupStatus{
				{
					Name:  "analytics",
					Phase: cloudv1.ClusterPhaseRunning,
					TiDB: &cloudv1.TiDBStatus{
						Members: []cloudv1.TiDBMember{
							{Name: "db-analytics-target-tidb-0"},
						},
					},
				},
			},
		},
	}

	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster, targetCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "replay-project",
		TenantId:  "tenant-1",
		Provider:  consts.ProviderTypeEKS,
		Region:    "us-west-2",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
			TenantId:    "tenant-1",
			ProjectId:   "source-project",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    consts.ProviderTypeEKS,
			Region:      "us-west-2",
			TenantId:    "tenant-1",
			ProjectId:   "target-project",
		},
	})

	replay := &cloudv1.Replay{ObjectMeta: metav1.ObjectMeta{Name: "r1"}}
	if err := builder.buildReplayTasks(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("buildReplayTasks() error = %v", err)
	}
	// The request does not match source/target placement, so current auto-fill defaults to target placement.
	// Target placement uses internal endpoints and keeps the valid analytics group.
	if len(replay.Spec.Tasks) != 2 {
		t.Fatalf("tasks len = %d, want 2 tasks", len(replay.Spec.Tasks))
	}
	if got := replay.Spec.Tasks[0].Name; got != replayTaskGroupDefaultName {
		t.Fatalf("tasks[0].Name = %q, want %q", got, replayTaskGroupDefaultName)
	}
	if got := replay.Spec.Tasks[0].TargetEndpoint; got != "db-tidb.tidbtarget-1.svc:4000" {
		t.Fatalf("tasks[0].TargetEndpoint = %q, want default internal endpoint", got)
	}
	if got := len(replay.Spec.Tasks[0].SourceAuditLogAddrs); got != 1 {
		t.Fatalf("len(tasks[0].SourceAuditLogAddrs) = %d, want 1 audit log", got)
	}
	if got := replay.Spec.Tasks[1].Name; got != "group-analytics" {
		t.Fatalf("tasks[1].Name = %q, want group-analytics", got)
	}
	if got := replay.Spec.Tasks[1].TargetEndpoint; got != "db-analytics-target-tidb.tidbtarget-1.svc:4000" {
		t.Fatalf("tasks[1].TargetEndpoint = %q, want analytics internal endpoint", got)
	}
	if got := len(replay.Spec.Tasks[1].SourceAuditLogAddrs); got != 1 {
		t.Fatalf("len(tasks[1].SourceAuditLogAddrs) = %d, want 1 audit log", got)
	}
}

func TestReplayBuilderBuildFailsWhenClusterTypesAreOmitted(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "source-1",
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.ProviderTypeLabel: "eks",
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://source-audit/prefix",
						RoleArn:   "arn:aws:iam::123456789012:role/replay",
					},
				},
			},
		},
	}
	targetCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "target-1",
			Namespace: namespace,
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Kubernetes: cloudv1.Kubernetes{
				Name:     "dp-target",
				Endpoint: "https://dp-target.example.com",
				Account:  "123456789012",
				Network:  "vpc-1",
				Provider: "eks",
				Version:  "1.29",
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster, targetCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-1",
		TenantId:  "tenant-1",
		Provider:  "eks",
		Region:    "us-west-2",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId: "source-1",
			Provider:  "eks",
			Region:    "us-west-2",
			TenantId:  "tenant-1",
			ProjectId: "project-1",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId: "target-1",
			Provider:  "eks",
			Region:    "us-west-2",
			TenantId:  "tenant-1",
			ProjectId: "project-1",
		},
	})

	_, err := builder.Build(t.Context())
	if err == nil {
		t.Fatal("Build() error = nil, want unsupported cluster type error")
	}
	if !strings.Contains(err.Error(), "unsupported cluster type CLUSTER_TYPE_UNSPECIFIED for replay") {
		t.Fatalf("Build() error = %v, want unsupported cluster type error", err)
	}
}

func TestReplayBuilderBuildReplayTasksFailsWhenTargetClusterTypeIsOmitted(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "project-1",
		TenantId:  "tenant-1",
		Provider:  "eks",
		Region:    "us-west-2",
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    "eks",
			Region:      "us-west-2",
			TenantId:    "tenant-1",
			ProjectId:   "project-1",
		},
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId: "target-1",
			Provider:  "eks",
			Region:    "us-west-2",
			TenantId:  "tenant-1",
			ProjectId: "project-1",
		},
	})

	// With the new design, unsupported cluster type is detected at context creation time.
	_, err := newReplayContext(builder.req, builder.cfg, builder.server)
	if err == nil {
		t.Fatal("newReplayContext() error = nil, want unsupported target cluster type error")
	}
	if !strings.Contains(err.Error(), "unsupported cluster type CLUSTER_TYPE_UNSPECIFIED for replay") {
		t.Fatalf("newReplayContext() error = %v, want unsupported target cluster type error", err)
	}
}

func TestReplayBuilderSetKubernetesInfoFromExplicitDataplaneRejectsConflictingKubeconfigInputs(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		Provider:  "ack",
		DataplaneKubernetes: &apiv1.Kubernetes{
			Name:       "dp-1",
			Endpoint:   "https://example.com:6443",
			Ca:         "fake-ca",
			AccountId:  "123",
			Vpc:        "vpc-1",
			Version:    "1.29",
			Kubeconfig: "apiVersion: v1",
			KubeConfigSecretRef: &apiv1.Kubernetes_SecretReference{
				Namespace: "ns-1",
				Name:      "existing-kubeconfig",
			},
			Info: &apiv1.Kubernetes_Alicloud_{
				Alicloud: &apiv1.Kubernetes_Alicloud{
					Vswitchs: []string{"vsw-1"},
				},
			},
		},
	})

	replay := &cloudv1.Replay{}
	err := builder.setKubernetesInfo(t.Context(), replay, mustNewReplayContext(t, builder))
	if err == nil {
		t.Fatalf("setExplicitKubernetesInfo() error = nil, want conflict error")
	}
	if !strings.Contains(err.Error(), "mutually exclusive") {
		t.Fatalf("setExplicitKubernetesInfo() error = %v, want mutually exclusive error", err)
	}
}

func TestReplayBuilderSetKubernetesInfoFromExplicitDataplaneCreateAllowsSecretRefWithoutProviderCloudConfig(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		Provider:  "ack",
		DataplaneKubernetes: &apiv1.Kubernetes{
			KubeConfigSecretRef: &apiv1.Kubernetes_SecretReference{
				Namespace: "ns-new",
				Name:      replayNewKubeConfig,
			},
		},
	})

	replay := &cloudv1.Replay{}
	if err := builder.setKubernetesInfo(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("setExplicitKubernetesInfo() error = %v", err)
	}
	if replay.Spec.Kubernetes.Provider != "ack" {
		t.Fatalf("setExplicitKubernetesInfo() provider = %q, want ack", replay.Spec.Kubernetes.Provider)
	}
	if replay.Spec.Kubernetes.KubeConfigSecretRef == nil ||
		replay.Spec.Kubernetes.KubeConfigSecretRef.Namespace != "ns-new" ||
		replay.Spec.Kubernetes.KubeConfigSecretRef.Name != replayNewKubeConfig {
		t.Fatalf("setExplicitKubernetesInfo() kubeconfig secret ref = %+v, want ns-new/new-kubeconfig", replay.Spec.Kubernetes.KubeConfigSecretRef)
	}
}

func TestReplayBuilderSetKubernetesInfoFromExplicitDataplaneAllowsEKSSecretRefWithoutProviderCloudConfig(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		Provider:  consts.ProviderTypeEKS,
		DataplaneKubernetes: &apiv1.Kubernetes{
			KubeConfigSecretRef: &apiv1.Kubernetes_SecretReference{
				Namespace: "ns-new",
				Name:      replayNewKubeConfig,
			},
		},
	})

	replay := &cloudv1.Replay{}
	if err := builder.setKubernetesInfo(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("setExplicitKubernetesInfo() error = %v", err)
	}
	if replay.Spec.Kubernetes.Provider != consts.ProviderTypeEKS {
		t.Fatalf("setExplicitKubernetesInfo() provider = %q, want %s", replay.Spec.Kubernetes.Provider, consts.ProviderTypeEKS)
	}
	if replay.Spec.Kubernetes.KubeConfigSecretRef == nil ||
		replay.Spec.Kubernetes.KubeConfigSecretRef.Namespace != "ns-new" ||
		replay.Spec.Kubernetes.KubeConfigSecretRef.Name != replayNewKubeConfig {
		t.Fatalf("setExplicitKubernetesInfo() kubeconfig secret ref = %+v, want ns-new/new-kubeconfig", replay.Spec.Kubernetes.KubeConfigSecretRef)
	}
	if replay.Spec.Kubernetes.Cloud.AWS != nil {
		t.Fatalf("setExplicitKubernetesInfo() cloud.aws = %+v, want nil", replay.Spec.Kubernetes.Cloud.AWS)
	}
}

func TestReplayBuilderSetKubernetesInfoFromExplicitDataplaneAllowsUnknownProviderWithoutProviderCloudConfig(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		Provider:  "gke",
		DataplaneKubernetes: &apiv1.Kubernetes{
			Name:     "dp-gke",
			Endpoint: "https://gke.example.com:6443",
		},
	})

	replay := &cloudv1.Replay{}
	if err := builder.setKubernetesInfo(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("setExplicitKubernetesInfo() error = %v", err)
	}
	if replay.Spec.Kubernetes.Provider != "gke" {
		t.Fatalf("setExplicitKubernetesInfo() provider = %q, want gke", replay.Spec.Kubernetes.Provider)
	}
	if !reflect.DeepEqual(replay.Spec.Kubernetes.Cloud, cloudv1.Cloud{}) {
		t.Fatalf("setExplicitKubernetesInfo() cloud = %+v, want empty", replay.Spec.Kubernetes.Cloud)
	}
}

func TestReplayBuilderSetKubernetesInfoFromExplicitDataplaneRejectsUnsupportedProviderCloudConfig(t *testing.T) {
	tests := []struct {
		name        string
		provider    string
		kubernetes  *apiv1.Kubernetes
		wantMessage string
	}{
		{
			name:     "unknown provider with AWS config",
			provider: "gke",
			kubernetes: &apiv1.Kubernetes{
				Info: &apiv1.Kubernetes_Aws_{
					Aws: &apiv1.Kubernetes_Aws{
						NodeSubnets: []*apiv1.Kubernetes_AwsSubnet{
							{Id: "subnet-new-node", AzName: "us-west-2a"},
						},
					},
				},
			},
			wantMessage: "unsupported cloud provider gke for replay",
		},
		{
			name:     "EKS provider with Alicloud config",
			provider: consts.ProviderTypeEKS,
			kubernetes: &apiv1.Kubernetes{
				Info: &apiv1.Kubernetes_Alicloud_{
					Alicloud: &apiv1.Kubernetes_Alicloud{
						Vswitchs: []string{"vsw-1"},
					},
				},
			},
			wantMessage: "unsupported cloud provider eks for replay",
		},
		{
			name:     "ACK provider with AWS config",
			provider: consts.ProviderTypeACK,
			kubernetes: &apiv1.Kubernetes{
				Info: &apiv1.Kubernetes_Aws_{
					Aws: &apiv1.Kubernetes_Aws{
						NodeSubnets: []*apiv1.Kubernetes_AwsSubnet{
							{Id: "subnet-new-node", AzName: "us-west-2a"},
						},
					},
				},
			},
			wantMessage: "unsupported cloud provider ack for replay",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			builder := newReplayBuilder(nil, &apiv1.Replay{
				ReplayId:            "r1",
				ProjectId:           "p1",
				Provider:            tt.provider,
				DataplaneKubernetes: tt.kubernetes,
			})
			err := builder.setKubernetesInfo(t.Context(), &cloudv1.Replay{}, mustNewReplayContext(t, builder))
			if err == nil {
				t.Fatal("setExplicitKubernetesInfo() error = nil, want unsupported provider cloud config")
			}
			if !strings.Contains(err.Error(), tt.wantMessage) {
				t.Fatalf("setExplicitKubernetesInfo() error = %v, want %q", err, tt.wantMessage)
			}
		})
	}
}

func TestReplayBuilderSetKubernetesInfoFromExplicitDataplaneAllowsPartialAWSSubnetInput(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		Provider:  consts.ProviderTypeEKS,
		DataplaneKubernetes: &apiv1.Kubernetes{
			Info: &apiv1.Kubernetes_Aws_{
				Aws: &apiv1.Kubernetes_Aws{
					NodeSubnets: []*apiv1.Kubernetes_AwsSubnet{
						{
							Id:     "subnet-new-node",
							AzName: "us-west-2a",
						},
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			Kubernetes: cloudv1.Kubernetes{
				Cloud: cloudv1.Cloud{
					AWS: &cloudv1.CloudAWS{},
				},
			},
		},
	}

	if err := builder.setKubernetesInfo(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("setExplicitKubernetesInfo() error = %v", err)
	}
	aws := replay.Spec.Kubernetes.Cloud.AWS
	if aws == nil {
		t.Fatal("setExplicitKubernetesInfo() cloud.aws = nil, want partial aws config")
	}
	if len(aws.SubnetGroups) != 1 {
		t.Fatalf("setExplicitKubernetesInfo() aws subnet groups = %+v, want node subnet group only", aws.SubnetGroups)
	}
	if aws.SubnetGroups[0].Name != cloudv1.AWSSubnetGroupNodes ||
		!reflect.DeepEqual(aws.SubnetGroups[0].Subnets, []cloudv1.AWSSubnet{{SubnetID: "subnet-new-node", Zone: "us-west-2a"}}) {
		t.Fatalf("setExplicitKubernetesInfo() node subnet group = %+v, want request node subnet only", aws.SubnetGroups[0])
	}
}

func TestReplayBuilderSetKubernetesInfoFromExplicitDataplaneAppliesAWSCloudConfig(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		Provider:  consts.ProviderTypeEKS,
		DataplaneKubernetes: &apiv1.Kubernetes{
			Info: &apiv1.Kubernetes_Aws_{
				Aws: &apiv1.Kubernetes_Aws{
					SecurityGroup: "sg-new",
					AdminNodeRole: "admin-new",
					NodeSubnets: []*apiv1.Kubernetes_AwsSubnet{
						{Id: "subnet-new-node", AzName: "us-west-2a"},
					},
					PublicLoadbalancerSubnets: []*apiv1.Kubernetes_AwsSubnet{
						{Id: "subnet-new-public", AzName: "us-west-2b"},
					},
					PrivateLoadbalancerSubnets: []*apiv1.Kubernetes_AwsSubnet{
						{Id: "subnet-new-private", AzName: "us-west-2c"},
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			Kubernetes: cloudv1.Kubernetes{
				Cloud: cloudv1.Cloud{
					AWS: &cloudv1.CloudAWS{
						EKSSecurityGroup: "sg-old",
						AdminNodeRole:    "admin-old",
						SubnetGroups: cloudv1.AWSSubnetGroups{
							{
								Name: cloudv1.AWSSubnetGroupNodes,
								Subnets: []cloudv1.AWSSubnet{
									{SubnetID: "subnet-old-node", Zone: "us-west-2a"},
								},
							},
							{
								Name: cloudv1.AWSSubnetGroupSecondaryCIDR,
								Subnets: []cloudv1.AWSSubnet{
									{SubnetID: "subnet-old-secondary", Zone: "us-west-2d"},
								},
							},
						},
					},
				},
			},
		},
	}

	if err := builder.setKubernetesInfo(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("setExplicitKubernetesInfo() error = %v", err)
	}

	if got := replay.Spec.Kubernetes.Provider; got != consts.ProviderTypeEKS {
		t.Fatalf("setExplicitKubernetesInfo() provider = %q, want %q", got, consts.ProviderTypeEKS)
	}
	aws := replay.Spec.Kubernetes.Cloud.AWS
	if aws == nil {
		t.Fatal("setExplicitKubernetesInfo() cloud aws = nil, want configured aws cloud")
	}
	if aws.EKSSecurityGroup != "sg-new" || aws.AdminNodeRole != "admin-new" {
		t.Fatalf("setExplicitKubernetesInfo() aws metadata = %+v, want new security group/admin role", aws)
	}
	if len(aws.SubnetGroups) != 3 {
		t.Fatalf("setExplicitKubernetesInfo() aws subnet groups = %+v, want only request subnet groups", aws.SubnetGroups)
	}
	_, subnets := aws.SubnetGroups.SubnetsFilterByGroupNameOrZones([]cloudv1.AWSSubnetGroupName{
		cloudv1.AWSSubnetGroupNodes,
		cloudv1.AWSSubnetGroupPublicNLB,
		cloudv1.AWSSubnetGroupPrivateNLB,
	}, nil)
	if !reflect.DeepEqual(subnets, []string{"subnet-new-node", "subnet-new-private", "subnet-new-public"}) {
		t.Fatalf("setExplicitKubernetesInfo() aws subnets = %+v, want new subnets", subnets)
	}
}

func TestReplayBuilderSetKubernetesInfoFromExplicitDataplaneOverridesProvidedFields(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		Provider:  "ack",
		DataplaneKubernetes: &apiv1.Kubernetes{
			Name:      replayNewDataplaneName,
			Endpoint:  "https://new.example.com:6443",
			Ca:        "new-ca",
			AccountId: "654321",
			Vpc:       "vpc-new",
			Version:   "1.30",
			Info: &apiv1.Kubernetes_Alicloud_{
				Alicloud: &apiv1.Kubernetes_Alicloud{
					Vswitchs: []string{"vsw-new"},
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			Kubernetes: cloudv1.Kubernetes{
				Name:     "dp-old",
				Endpoint: "https://old.example.com:6443",
				CAData:   []byte("old-ca"),
				Account:  "123456",
				Network:  "vpc-old",
				Provider: "ack",
				Version:  "1.29",
				Cloud: cloudv1.Cloud{
					Ack: &cloudv1.CloudAck{
						Subnets: []string{"vsw-old"},
					},
				},
			},
		},
	}

	if err := builder.setKubernetesInfo(t.Context(), replay, mustNewReplayContext(t, builder)); err != nil {
		t.Fatalf("setExplicitKubernetesInfo() error = %v", err)
	}

	got := replay.Spec.Kubernetes
	if got.Name != replayNewDataplaneName || got.Endpoint != "https://new.example.com:6443" ||
		string(got.CAData) != "new-ca" || got.Account != "654321" ||
		got.Network != "vpc-new" || got.Version != "1.30" {
		t.Fatalf("setExplicitKubernetesInfo() did not override provided metadata: %+v", got)
	}
	if got.Cloud.Ack == nil || !reflect.DeepEqual(got.Cloud.Ack.Subnets, []string{"vsw-new"}) {
		t.Fatalf("setExplicitKubernetesInfo() cloud ack = %+v, want [vsw-new]", got.Cloud.Ack)
	}
}

func TestReplayBuilderSetReplayConfigImageBehavior(t *testing.T) {
	tests := []struct {
		name      string
		req       *apiv1.Replay
		replay    *cloudv1.Replay
		wantImage string
	}{
		{
			name: "explicit image overrides existing image",
			req: &apiv1.Replay{
				ReplayId:  "r1",
				ProjectId: "p1",
				Defaults: &apiv1.Replay_ReplayDefaults{
					Image: ptr.To("custom/replayer:v2"),
				},
			},
			replay: &cloudv1.Replay{
				Spec: cloudv1.ReplaySpec{
					ReplayConfig: cloudv1.ReplayConfig{Image: "old/replayer:v1"},
				},
			},
			wantImage: "custom/replayer:v2",
		},
		{
			name: "defaults backfill system image",
			req: &apiv1.Replay{
				ReplayId:  "r1",
				ProjectId: "p1",
				Defaults: &apiv1.Replay_ReplayDefaults{
					Replicas: ptr.To(int32(2)),
				},
			},
			replay:    &cloudv1.Replay{},
			wantImage: cvt.DefaultReplayImage,
		},
		{
			name: "auto-task defaults also backfill system image",
			req: &apiv1.Replay{
				ReplayId:              "r1",
				ProjectId:             "p1",
				PremiumTargetEndpoint: ptr.To("tidb.premium.example.com:4000"),
				Defaults: &apiv1.Replay_ReplayDefaults{
					SourceAuditLog: &apiv1.Replay_AuditLogAddr{
						Source: &apiv1.Replay_AuditLogAddr_Aliyun{
							Aliyun: &apiv1.Replay_AliyunAuditLog{
								BucketUrl: "s3://bucket/prefix?endpoint=http://oss-ap-southeast-1.aliyuncs.com",
							},
						},
					},
				},
			},
			replay:    &cloudv1.Replay{},
			wantImage: cvt.DefaultReplayImage,
		},
		{
			name: "without defaults existing image is preserved",
			req: &apiv1.Replay{
				ReplayId:  "r1",
				ProjectId: "p1",
			},
			replay: &cloudv1.Replay{
				Spec: cloudv1.ReplaySpec{
					ReplayConfig: cloudv1.ReplayConfig{Image: "old/replayer:v1"},
				},
			},
			wantImage: "old/replayer:v1",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			builder := newReplayBuilder(nil, tt.req)
			builder.setReplayConfig(tt.replay)

			if got := tt.replay.Spec.ReplayConfig.Image; got != tt.wantImage {
				t.Fatalf("setReplayConfig() image = %q, want %q", got, tt.wantImage)
			}
		})
	}
}

func TestReplayBuilderBuildForUpdateAllowsDataplaneKubernetesChangesForDedicatedReplay(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:      "r1",
		ProjectId:     "p1",
		Provider:      "ack",
		SourceCluster: dedicatedReplayClusterRef("source-1", "p1", consts.ProviderTypeACK, "ap-southeast-1"),
		TargetCluster: dedicatedReplayClusterRef("target-1", "p1", consts.ProviderTypeACK, "ap-southeast-1"),
		Tasks: []*apiv1.Replay_ReplayTask{
			{
				Name: replayTaskGroup0Name,
				Source: &apiv1.Replay_DataSource{
					AuditLogs: []*apiv1.Replay_AuditLogAddr{
						{
							Source: &apiv1.Replay_AuditLogAddr_Aliyun{
								Aliyun: &apiv1.Replay_AliyunAuditLog{
									BucketUrl: "s3://bucket/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
								},
							},
						},
					},
				},
				Target: &apiv1.Replay_DataTarget{
					Endpoint: "tidb.example.com:4000",
				},
			},
		},
		DataplaneKubernetes: &apiv1.Kubernetes{
			Name:      replayNewDataplaneName,
			Endpoint:  "https://new.example.com:6443",
			Ca:        "new-ca",
			AccountId: "654321",
			Vpc:       "vpc-new",
			Version:   "1.30",
			Info: &apiv1.Kubernetes_Alicloud_{
				Alicloud: &apiv1.Kubernetes_Alicloud{
					Vswitchs: []string{"vsw-new"},
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			TargetClusterType: cloudv1.ReplayClusterTypeDedicated,
		},
	}

	if err := builder.BuildForUpdate(t.Context(), replay); err != nil {
		t.Fatalf("BuildForUpdate() error = %v", err)
	}

	if got := replay.Spec.Kubernetes.Name; got != replayNewDataplaneName {
		t.Fatalf("BuildForUpdate() kubernetes.name = %q, want %q", got, replayNewDataplaneName)
	}
	if replay.Spec.Kubernetes.Cloud.Ack == nil || !reflect.DeepEqual(replay.Spec.Kubernetes.Cloud.Ack.Subnets, []string{"vsw-new"}) {
		t.Fatalf("BuildForUpdate() kubernetes.cloud.ack = %+v, want [vsw-new]", replay.Spec.Kubernetes.Cloud.Ack)
	}
}

func TestReplayBuilderBuildForUpdateDedicatedDataplaneKubernetesReplacesExistingDataplane(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:      "r1",
		ProjectId:     "p1",
		Provider:      "ack",
		SourceCluster: dedicatedReplayClusterRef("source-1", "p1", consts.ProviderTypeACK, "ap-southeast-1"),
		TargetCluster: dedicatedReplayClusterRef("target-1", "p1", consts.ProviderTypeACK, "ap-southeast-1"),
		Tasks: []*apiv1.Replay_ReplayTask{
			{
				Name: replayTaskGroup0Name,
				Source: &apiv1.Replay_DataSource{
					AuditLogs: []*apiv1.Replay_AuditLogAddr{
						{
							Source: &apiv1.Replay_AuditLogAddr_Aliyun{
								Aliyun: &apiv1.Replay_AliyunAuditLog{
									BucketUrl: "s3://bucket/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
								},
							},
						},
					},
				},
				Target: &apiv1.Replay_DataTarget{
					Endpoint: "tidb.example.com:4000",
				},
			},
		},
		DataplaneKubernetes: &apiv1.Kubernetes{
			Info: &apiv1.Kubernetes_Alicloud_{
				Alicloud: &apiv1.Kubernetes_Alicloud{
					Vswitchs: []string{"vsw-new"},
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			TargetClusterType: cloudv1.ReplayClusterTypeDedicated,
			Kubernetes: cloudv1.Kubernetes{
				Name:     "dp-old",
				Endpoint: "https://old.example.com:6443",
				CAData:   []byte("old-ca"),
				Account:  "123456",
				Network:  "vpc-old",
				Provider: consts.ProviderTypeACK,
				Version:  "1.28",
				Cloud: cloudv1.Cloud{
					Ack: &cloudv1.CloudAck{
						Subnets: []string{"vsw-old"},
					},
				},
			},
		},
	}

	if err := builder.BuildForUpdate(t.Context(), replay); err != nil {
		t.Fatalf("BuildForUpdate() error = %v", err)
	}

	if replay.Spec.Kubernetes.Name != "" ||
		replay.Spec.Kubernetes.Endpoint != "" ||
		len(replay.Spec.Kubernetes.CAData) != 0 ||
		replay.Spec.Kubernetes.Account != "" ||
		replay.Spec.Kubernetes.Network != "" ||
		replay.Spec.Kubernetes.Provider != consts.ProviderTypeACK ||
		replay.Spec.Kubernetes.Version != "" {
		t.Fatalf("BuildForUpdate() did not replace existing kubernetes metadata: %+v", replay.Spec.Kubernetes)
	}
	if replay.Spec.Kubernetes.Cloud.Ack == nil || !reflect.DeepEqual(replay.Spec.Kubernetes.Cloud.Ack.Subnets, []string{"vsw-new"}) {
		t.Fatalf("BuildForUpdate() kubernetes.cloud.ack = %+v, want [vsw-new]", replay.Spec.Kubernetes.Cloud.Ack)
	}
}

func TestReplayBuilderBuildForUpdatePremiumWithoutExplicitDataplaneRejectsIndividualPlacement(t *testing.T) {
	origFn := newReplayClusterProviderFn
	t.Cleanup(func() { newReplayClusterProviderFn = origFn })
	newReplayClusterProviderFn = func(_ *replayContext, ref *apiv1.Replay_ClusterRef) (replayClusterProvider, error) {
		return &replayClusterProviderStub{
			clusterType: ref.ClusterType,
			resolveDataplane: func(context.Context) (replayDataplaneResolution, error) {
				return replayDataplaneResolution{}, errReplayDataplaneUnsupportedPlacement
			},
		}, nil
	}

	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:      "r1",
		ProjectId:     "p1",
		Provider:      "ack",
		SourceCluster: premiumReplayClusterRef("source-premium", "p1", consts.ProviderTypeACK, "ap-southeast-1"),
		TargetCluster: premiumReplayClusterRef("target-premium", "p1", consts.ProviderTypeACK, "ap-southeast-1"),
		Tasks: []*apiv1.Replay_ReplayTask{
			{
				Name: replayTaskGroup0Name,
				Source: &apiv1.Replay_DataSource{
					AuditLogs: []*apiv1.Replay_AuditLogAddr{
						{
							Source: &apiv1.Replay_AuditLogAddr_Aliyun{
								Aliyun: &apiv1.Replay_AliyunAuditLog{
									BucketUrl: "s3://bucket/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
								},
							},
						},
					},
				},
				Target: &apiv1.Replay_DataTarget{
					Endpoint: "tidb.example.com:4000",
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			TargetClusterType: cloudv1.ReplayClusterTypePremium,
			SourceClusterID:   ptr.To("source-premium"),
			SourceClusterType: ptr.To(cloudv1.ReplayClusterTypePremium),
			SourceProvider:    ptr.To(consts.ProviderTypeACK),
			SourceRegion:      ptr.To("ap-southeast-1"),
			Kubernetes: cloudv1.Kubernetes{
				Name:     "dp-old",
				Endpoint: "https://old.example.com:6443",
				CAData:   []byte("old-ca"),
				Account:  "123456",
				Network:  "vpc-old",
				Provider: consts.ProviderTypeACK,
				Version:  "1.28",
			},
		},
	}

	err := builder.BuildForUpdate(t.Context(), replay)
	if err == nil {
		t.Fatal("BuildForUpdate() error = nil, want deploy position auto-fill error")
	}
	if !strings.Contains(err.Error(), "not supported deploy position for filling dataplane info automatically") {
		t.Fatalf("BuildForUpdate() error = %v, want deploy position auto-fill error", err)
	}
}

func TestReplayBuilderBuildForUpdatePremiumNoopRejectsMissingRebuildMaterial(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		Provider:  consts.ProviderTypeACK,
		Region:    "ap-southeast-1",
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			State:             cloudv1.ReplayStateDeploy,
			TargetClusterType: cloudv1.ReplayClusterTypePremium,
			TargetClusterID:   "target-premium",
			TargetProvider:    consts.ProviderTypeACK,
			TargetRegion:      "ap-southeast-1",
			Kubernetes: cloudv1.Kubernetes{
				Name:     "dp-old",
				Endpoint: "https://old.example.com:6443",
				Provider: consts.ProviderTypeACK,
			},
			Tasks: []cloudv1.ReplayTask{
				{
					Name:           replayTaskGroup0Name,
					TargetEndpoint: "tidb.old.example.com:4000",
				},
			},
		},
	}

	if err := assertBuildForUpdateRejectedWithoutMutatingSpec(t, builder, replay); err == nil {
		t.Fatal("assertBuildForUpdateRejectedWithoutMutatingSpec() error = nil, want rejected update")
	}
}

func TestReplayBuilderBuildForUpdateRefreshesDedicatedDataplaneKubernetesWithoutExplicitDataplane(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "p1")
	targetCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "target-1",
			Namespace: namespace,
		},
		Spec: cloudv1.ClusterSpec{
			Kubernetes: cloudv1.Kubernetes{
				Name:     replayNewDataplaneName,
				Endpoint: "https://new.example.com:6443",
				CAData:   []byte("new-ca"),
				Account:  "654321",
				Network:  "vpc-new",
				Provider: consts.ProviderTypeACK,
				Version:  "1.30",
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(targetCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:      "r1",
		TenantId:      "tenant-1",
		ProjectId:     "p1",
		Provider:      consts.ProviderTypeACK,
		Region:        "ap-southeast-1",
		SourceCluster: dedicatedReplayClusterRef("source-1", "p1", consts.ProviderTypeACK, "ap-southeast-1"),
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-1",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			TenantId:    "tenant-1",
			ProjectId:   "p1",
			Provider:    consts.ProviderTypeACK,
			Region:      "ap-southeast-1",
		},
		Tasks: []*apiv1.Replay_ReplayTask{
			{
				Name: replayTaskGroup0Name,
				Source: &apiv1.Replay_DataSource{
					AuditLogs: []*apiv1.Replay_AuditLogAddr{
						{
							Source: &apiv1.Replay_AuditLogAddr_Aliyun{
								Aliyun: &apiv1.Replay_AliyunAuditLog{
									BucketUrl: "s3://bucket/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
								},
							},
						},
					},
				},
				Target: &apiv1.Replay_DataTarget{
					Endpoint: "tidb.example.com:4000",
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			TargetClusterID:   "target-1",
			TargetClusterType: cloudv1.ReplayClusterTypeDedicated,
			TargetProvider:    consts.ProviderTypeACK,
			TargetRegion:      "ap-southeast-1",
			Kubernetes: cloudv1.Kubernetes{
				Name:     "dp-old",
				Endpoint: "https://old.example.com:6443",
				CAData:   []byte("old-ca"),
				Account:  "123456",
				Network:  "vpc-old",
				Provider: consts.ProviderTypeACK,
				Version:  "1.28",
			},
		},
	}

	if err := builder.BuildForUpdate(t.Context(), replay); err != nil {
		t.Fatalf("BuildForUpdate() error = %v", err)
	}

	if replay.Spec.Kubernetes.Name != replayNewDataplaneName ||
		replay.Spec.Kubernetes.Endpoint != "https://new.example.com:6443" ||
		string(replay.Spec.Kubernetes.CAData) != "new-ca" ||
		replay.Spec.Kubernetes.Account != "654321" ||
		replay.Spec.Kubernetes.Network != "vpc-new" ||
		replay.Spec.Kubernetes.Provider != consts.ProviderTypeACK ||
		replay.Spec.Kubernetes.Version != "1.30" {
		t.Fatalf("BuildForUpdate() did not refresh dedicated dataplane kubernetes: %+v", replay.Spec.Kubernetes)
	}
}

func TestReplayBuilderBuildForUpdatePremiumDataplaneOnlyUpdateRejectsMissingRebuildMaterial(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		Provider:  "ack",
		DataplaneKubernetes: &apiv1.Kubernetes{
			Name: "dp-1",
			Info: &apiv1.Kubernetes_Alicloud_{
				Alicloud: &apiv1.Kubernetes_Alicloud{
					Vswitchs: []string{"vsw-1"},
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			TargetClusterID:   "target-premium",
			TargetClusterType: cloudv1.ReplayClusterTypePremium,
			TargetProvider:    "ack",
			TargetRegion:      "ap-southeast-1",
			SourceClusterID:   ptr.To("source-premium"),
			SourceClusterType: ptr.To(cloudv1.ReplayClusterTypePremium),
			SourceProvider:    ptr.To("ack"),
			SourceRegion:      ptr.To("ap-southeast-1"),
			Tasks: []cloudv1.ReplayTask{
				{
					Name:           replayTaskGroup0Name,
					TargetEndpoint: "tidb.example.com:4000",
					SourceAuditLogAddrs: []cloudv1.AuditLogAddr{
						{
							Aliyun: &cloudv1.AliyunAuditLog{
								BucketURL: "s3://bucket/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
							},
						},
					},
				},
			},
		},
	}

	if err := assertBuildForUpdateRejectedWithoutMutatingSpec(t, builder, replay); err == nil {
		t.Fatal("assertBuildForUpdateRejectedWithoutMutatingSpec() error = nil, want rejected update")
	}
}

func TestReplayServerCreateKubeConfigSecretFromDataOverwritesExistingSecret(t *testing.T) {
	existingSecret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "r1-kubeconfig",
			Namespace: "replay-ns",
		},
		Data: map[string][]byte{
			"kubeconfig": []byte("old-kubeconfig"),
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(existingSecret).
		Build()
	server := NewReplayServer(kubeClient)

	replay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "r1",
			Namespace: "replay-ns",
		},
	}

	if err := server.createKubeConfigSecretFromData(t.Context(), replay, []byte(replayNewKubeConfig)); err != nil {
		t.Fatalf("createKubeConfigSecretFromData() error = %v", err)
	}

	updatedSecret := &corev1.Secret{}
	if err := kubeClient.Get(t.Context(), client.ObjectKey{
		Namespace: existingSecret.Namespace,
		Name:      existingSecret.Name,
	}, updatedSecret); err != nil {
		t.Fatalf("Get(secret) error = %v", err)
	}
	if got := string(updatedSecret.Data["kubeconfig"]); got != replayNewKubeConfig {
		t.Fatalf("secret kubeconfig = %q, want %q", got, replayNewKubeConfig)
	}
}

func TestReplayServerCreateReplayDuplicateDoesNotOverwriteGeneratedKubeconfigSecret(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	existingReplay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "replay-1",
			Namespace: namespace,
		},
	}
	existingSecret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "replay-1-kubeconfig",
			Namespace: namespace,
		},
		Data: map[string][]byte{
			"kubeconfig": []byte("old-kubeconfig"),
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(existingReplay, existingSecret).
		Build()
	server := NewReplayServer(kubeClient)

	req := &apiv1.CreateReplayRequest{
		Replay: &apiv1.Replay{
			ReplayId:  "replay-1",
			TenantId:  "tenant-1",
			ProjectId: "project-1",
			Provider:  "ack",
			Region:    "ap-southeast-1",
			State:     apiv1.Replay_STATE_DEPLOY,
			TargetCluster: &apiv1.Replay_ClusterRef{
				ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
				ClusterId:   "target-premium",
				Provider:    "ack",
				Region:      "ap-southeast-1",
				TenantId:    "tenant-1",
				ProjectId:   "project-1",
			},
			DataplaneKubernetes: &apiv1.Kubernetes{
				Name:       "dp-1",
				Version:    "1.29",
				Endpoint:   "https://example.com:6443",
				Ca:         "fake-ca",
				AccountId:  "123456",
				Vpc:        "vpc-1",
				Kubeconfig: replayNewKubeConfig,
				Info: &apiv1.Kubernetes_Alicloud_{
					Alicloud: &apiv1.Kubernetes_Alicloud{
						Vswitchs: []string{"vsw-1"},
					},
				},
			},
			Tasks: []*apiv1.Replay_ReplayTask{
				{
					Name: replayTaskGroup0Name,
					Source: &apiv1.Replay_DataSource{
						AuditLogs: []*apiv1.Replay_AuditLogAddr{
							{
								Source: &apiv1.Replay_AuditLogAddr_Aliyun{
									Aliyun: &apiv1.Replay_AliyunAuditLog{
										BucketUrl: "s3://bucket/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
									},
								},
							},
						},
					},
					Target: &apiv1.Replay_DataTarget{
						Endpoint: "tidb.example.com:4000",
					},
				},
			},
		},
	}

	_, err := server.CreateReplay(t.Context(), req)
	if err == nil {
		t.Fatal("CreateReplay() error = nil, want AlreadyExists")
	}
	if status.Code(err) != codes.AlreadyExists {
		t.Fatalf("CreateReplay() code = %v, want %v", status.Code(err), codes.AlreadyExists)
	}

	updatedSecret := &corev1.Secret{}
	if err := kubeClient.Get(t.Context(), client.ObjectKey{
		Namespace: existingSecret.Namespace,
		Name:      existingSecret.Name,
	}, updatedSecret); err != nil {
		t.Fatalf("Get(secret) error = %v", err)
	}
	if got := string(updatedSecret.Data["kubeconfig"]); got != "old-kubeconfig" {
		t.Fatalf("secret kubeconfig = %q, want %q", got, "old-kubeconfig")
	}
}

func TestReplayServerCreateReplayFailureCleansGeneratedKubeconfigSecret(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithInterceptorFuncs(interceptor.Funcs{
			Create: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.CreateOption) error {
				if _, ok := obj.(*cloudv1.Replay); ok {
					return fmt.Errorf("inject replay create failure")
				}
				return c.Create(ctx, obj, opts...)
			},
		}).
		Build()
	server := NewReplayServer(kubeClient)

	req := &apiv1.CreateReplayRequest{
		Replay: &apiv1.Replay{
			ReplayId:  "replay-1",
			TenantId:  "tenant-1",
			ProjectId: "project-1",
			Provider:  "ack",
			Region:    "ap-southeast-1",
			State:     apiv1.Replay_STATE_DEPLOY,
			TargetCluster: &apiv1.Replay_ClusterRef{
				ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
				ClusterId:   "target-premium",
				Provider:    "ack",
				Region:      "ap-southeast-1",
				TenantId:    "tenant-1",
				ProjectId:   "project-1",
			},
			DataplaneKubernetes: &apiv1.Kubernetes{
				Name:       "dp-1",
				Version:    "1.29",
				Endpoint:   "https://example.com:6443",
				Ca:         "fake-ca",
				AccountId:  "123456",
				Vpc:        "vpc-1",
				Kubeconfig: replayNewKubeConfig,
				Info: &apiv1.Kubernetes_Alicloud_{
					Alicloud: &apiv1.Kubernetes_Alicloud{
						Vswitchs: []string{"vsw-1"},
					},
				},
			},
			Tasks: []*apiv1.Replay_ReplayTask{
				{
					Name: replayTaskGroup0Name,
					Source: &apiv1.Replay_DataSource{
						AuditLogs: []*apiv1.Replay_AuditLogAddr{
							{
								Source: &apiv1.Replay_AuditLogAddr_Aliyun{
									Aliyun: &apiv1.Replay_AliyunAuditLog{
										BucketUrl: "s3://bucket/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
									},
								},
							},
						},
					},
					Target: &apiv1.Replay_DataTarget{
						Endpoint: "tidb.example.com:4000",
					},
				},
			},
		},
	}

	_, err := server.CreateReplay(t.Context(), req)
	if err == nil {
		t.Fatal("CreateReplay() error = nil, want injected failure")
	}
	if !strings.Contains(err.Error(), "inject replay create failure") {
		t.Fatalf("CreateReplay() error = %v, want injected failure", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	secret := &corev1.Secret{}
	err = kubeClient.Get(t.Context(), client.ObjectKey{
		Namespace: namespace,
		Name:      "replay-1-kubeconfig",
	}, secret)
	if !apierrors.IsNotFound(err) {
		t.Fatalf("Get(secret) error = %v, want not found", err)
	}
}

func TestReplayServerCreateReplayFailureKeepsExplicitKubeconfigSecret(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	existingSecret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "external-kubeconfig",
			Namespace: "user-ns",
		},
		Data: map[string][]byte{
			"kubeconfig": []byte("external-kubeconfig-data"),
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(existingSecret).
		WithInterceptorFuncs(interceptor.Funcs{
			Create: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.CreateOption) error {
				if _, ok := obj.(*cloudv1.Replay); ok {
					return fmt.Errorf("inject replay create failure")
				}
				return c.Create(ctx, obj, opts...)
			},
		}).
		Build()
	server := NewReplayServer(kubeClient)

	req := &apiv1.CreateReplayRequest{
		Replay: &apiv1.Replay{
			ReplayId:  "replay-1",
			TenantId:  "tenant-1",
			ProjectId: "project-1",
			Provider:  "ack",
			Region:    "ap-southeast-1",
			State:     apiv1.Replay_STATE_DEPLOY,
			TargetCluster: &apiv1.Replay_ClusterRef{
				ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
				ClusterId:   "target-premium",
				Provider:    "ack",
				Region:      "ap-southeast-1",
				TenantId:    "tenant-1",
				ProjectId:   "project-1",
			},
			DataplaneKubernetes: &apiv1.Kubernetes{
				Name: "dp-1",
				KubeConfigSecretRef: &apiv1.Kubernetes_SecretReference{
					Namespace: "user-ns",
					Name:      "external-kubeconfig",
				},
			},
			Tasks: []*apiv1.Replay_ReplayTask{
				{
					Name: replayTaskGroup0Name,
					Source: &apiv1.Replay_DataSource{
						AuditLogs: []*apiv1.Replay_AuditLogAddr{
							{
								Source: &apiv1.Replay_AuditLogAddr_Aliyun{
									Aliyun: &apiv1.Replay_AliyunAuditLog{
										BucketUrl: "s3://bucket/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
									},
								},
							},
						},
					},
					Target: &apiv1.Replay_DataTarget{
						Endpoint: "tidb.example.com:4000",
					},
				},
			},
		},
	}

	_, err := server.CreateReplay(t.Context(), req)
	if err == nil {
		t.Fatal("CreateReplay() error = nil, want injected failure")
	}
	if !strings.Contains(err.Error(), "inject replay create failure") {
		t.Fatalf("CreateReplay() error = %v, want injected failure", err)
	}

	secret := &corev1.Secret{}
	err = kubeClient.Get(t.Context(), client.ObjectKey{
		Namespace: existingSecret.Namespace,
		Name:      existingSecret.Name,
	}, secret)
	if err != nil {
		t.Fatalf("Get(secret) error = %v, want existing external secret", err)
	}
	if got := string(secret.Data["kubeconfig"]); got != "external-kubeconfig-data" {
		t.Fatalf("secret kubeconfig = %q, want %q", got, "external-kubeconfig-data")
	}
	replay := &cloudv1.Replay{}
	if kubeClient.Get(t.Context(), client.ObjectKey{Name: "replay-1", Namespace: namespace}, replay) == nil {
		t.Fatalf("Replay unexpectedly created: %+v", replay)
	}
}

func TestReplayServerCreatePasswordSecretPasswordOnlyUpdateIsRejected(t *testing.T) {
	existingSecret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "replay-r1-target-db-password",
			Namespace: "replay-ns",
		},
		Data: map[string][]byte{
			"old-root": []byte("old-pass"),
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(existingSecret).
		Build()
	server := NewReplayServer(kubeClient)

	replay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "r1",
			Namespace: "replay-ns",
		},
		Spec: cloudv1.ReplaySpec{
			PasswordSecretRef: corev1.SecretReference{
				Name:      existingSecret.Name,
				Namespace: existingSecret.Namespace,
			},
			Tasks: []cloudv1.ReplayTask{
				{Username: "old-root"},
			},
		},
	}
	req := &apiv1.Replay{
		Defaults: &apiv1.Replay_ReplayDefaults{
			Credential: &apiv1.Replay_DatabaseCredential{
				Password: "new-pass",
			},
		},
	}

	err := server.createPasswordSecret(t.Context(), replay, req)
	if err == nil {
		t.Fatal("createPasswordSecret() error = nil, want invalid argument")
	}
	if status.Code(err) != codes.InvalidArgument {
		t.Fatalf("createPasswordSecret() code = %v, want %v", status.Code(err), codes.InvalidArgument)
	}
}

func TestReplayServerCreatePasswordSecretUsernameOnlyUpdateIsRejected(t *testing.T) {
	existingSecret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "replay-r1-target-db-password",
			Namespace: "replay-ns",
		},
		Data: map[string][]byte{
			"old-root": []byte("old-pass"),
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(existingSecret).
		Build()
	server := NewReplayServer(kubeClient)

	replay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "r1",
			Namespace: "replay-ns",
		},
		Spec: cloudv1.ReplaySpec{
			PasswordSecretRef: corev1.SecretReference{
				Name:      existingSecret.Name,
				Namespace: existingSecret.Namespace,
			},
			Tasks: []cloudv1.ReplayTask{
				{Username: "old-root"},
			},
		},
	}
	replay.Spec.Tasks[0].Username = replayNewRootUsername
	req := &apiv1.Replay{
		Defaults: &apiv1.Replay_ReplayDefaults{
			Credential: &apiv1.Replay_DatabaseCredential{
				Username: replayNewRootUsername,
			},
		},
	}

	err := server.createPasswordSecret(t.Context(), replay, req)
	if err == nil {
		t.Fatal("createPasswordSecret() error = nil, want invalid argument")
	}
	if status.Code(err) != codes.InvalidArgument {
		t.Fatalf("createPasswordSecret() code = %v, want %v", status.Code(err), codes.InvalidArgument)
	}
}

func TestReplayServerCreatePasswordSecretUsernameOnlyCreateIsRejected(t *testing.T) {
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		Build()
	server := NewReplayServer(kubeClient)

	replay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "r1",
			Namespace: "replay-ns",
		},
		Spec: cloudv1.ReplaySpec{
			Tasks: []cloudv1.ReplayTask{
				{Username: replayNewRootUsername},
			},
		},
	}
	req := &apiv1.Replay{
		Defaults: &apiv1.Replay_ReplayDefaults{
			Credential: &apiv1.Replay_DatabaseCredential{
				Username: replayNewRootUsername,
			},
		},
	}

	err := server.createPasswordSecret(t.Context(), replay, req)
	if err == nil {
		t.Fatal("createPasswordSecret() error = nil, want invalid argument")
	}
	if status.Code(err) != codes.InvalidArgument {
		t.Fatalf("createPasswordSecret() code = %v, want %v", status.Code(err), codes.InvalidArgument)
	}
}

func TestReplayServerUpdateReplayNilRequest(t *testing.T) {
	server := NewReplayServer(nil)

	_, err := server.UpdateReplay(t.Context(), nil)
	if err == nil {
		t.Fatalf("UpdateReplay(nil) error = nil, want invalid argument")
	}
	if status.Code(err) != codes.InvalidArgument {
		t.Fatalf("UpdateReplay(nil) code = %v, want %v", status.Code(err), codes.InvalidArgument)
	}

	_, err = server.UpdateReplay(t.Context(), &apiv1.UpdateReplayRequest{})
	if err == nil {
		t.Fatalf("UpdateReplay(empty) error = nil, want invalid argument")
	}
	if status.Code(err) != codes.InvalidArgument {
		t.Fatalf("UpdateReplay(empty) code = %v, want %v", status.Code(err), codes.InvalidArgument)
	}
}

func TestReplayServerUpdateReplayPremiumShapeErrorsReturnInternal(t *testing.T) {
	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")

	tests := []struct {
		name   string
		req    *apiv1.Replay
		replay *cloudv1.Replay
		want   string
	}{
		{
			name: "missing premium target endpoint",
			req: &apiv1.Replay{
				ReplayId:  "replay-1",
				ProjectId: "project-1",
				State:     apiv1.Replay_STATE_DEPLOY,
				SourceCluster: &apiv1.Replay_ClusterRef{
					ClusterId:   "source-premium",
					ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
					ProjectId:   "project-1",
					Provider:    "ack",
					Region:      "ap-southeast-1",
				},
				TargetCluster: &apiv1.Replay_ClusterRef{
					ClusterId:   "target-premium",
					ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
					Provider:    "ack",
					Region:      "ap-southeast-1",
				},
				Defaults: &apiv1.Replay_ReplayDefaults{
					SourceAuditLog: &apiv1.Replay_AuditLogAddr{
						Source: &apiv1.Replay_AuditLogAddr_Aliyun{
							Aliyun: &apiv1.Replay_AliyunAuditLog{
								BucketUrl: "s3://bucket/prefix",
							},
						},
					},
				},
			},
			replay: &cloudv1.Replay{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "replay-1",
					Namespace: namespace,
				},
				Spec: cloudv1.ReplaySpec{
					TargetClusterID:   "target-premium",
					State:             cloudv1.ReplayStateDeploy,
					TargetClusterType: cloudv1.ReplayClusterTypePremium,
					TargetProvider:    "ack",
					TargetRegion:      "ap-southeast-1",
					SourceClusterID:   ptr.To("source-premium"),
					SourceClusterType: ptr.To(cloudv1.ReplayClusterTypePremium),
					SourceProvider:    ptr.To("ack"),
					SourceRegion:      ptr.To("ap-southeast-1"),
				},
			},
			want: "target endpoint is not set",
		},
		{
			name: "missing source audit log",
			req: &apiv1.Replay{
				ReplayId:              "replay-1",
				ProjectId:             "project-1",
				State:                 apiv1.Replay_STATE_DEPLOY,
				PremiumTargetEndpoint: ptr.To("tidb.example.com:4000"),
				SourceCluster: &apiv1.Replay_ClusterRef{
					ClusterId:   "source-premium",
					ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
					ProjectId:   "project-1",
					Provider:    "ack",
					Region:      "ap-southeast-1",
				},
				TargetCluster: &apiv1.Replay_ClusterRef{
					ClusterId:   "target-premium",
					ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
					Provider:    "ack",
					Region:      "ap-southeast-1",
				},
			},
			replay: &cloudv1.Replay{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "replay-1",
					Namespace: namespace,
				},
				Spec: cloudv1.ReplaySpec{
					TargetClusterID:   "target-premium",
					State:             cloudv1.ReplayStateDeploy,
					TargetClusterType: cloudv1.ReplayClusterTypePremium,
					TargetProvider:    "ack",
					TargetRegion:      "ap-southeast-1",
					SourceClusterID:   ptr.To("source-premium"),
					SourceClusterType: ptr.To(cloudv1.ReplayClusterTypePremium),
					SourceProvider:    ptr.To("ack"),
					SourceRegion:      ptr.To("ap-southeast-1"),
				},
			},
			want: "source audit log is not set for premium replay",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
				t.Fatalf("AddToScheme() error = %v", err)
			}
			kubeClient := fake.NewClientBuilder().WithScheme(scheme.Scheme).WithObjects(tt.replay).Build()
			server := NewReplayServer(kubeClient)

			_, err := server.UpdateReplay(t.Context(), &apiv1.UpdateReplayRequest{Replay: tt.req})
			if err == nil {
				t.Fatal("UpdateReplay() error = nil, want InvalidArgument")
			}
			if status.Code(err) != codes.InvalidArgument {
				t.Fatalf("UpdateReplay() code = %v, want %v", status.Code(err), codes.InvalidArgument)
			}
			if !strings.Contains(err.Error(), tt.want) {
				t.Fatalf("UpdateReplay() error = %v, want substring %q", err, tt.want)
			}
		})
	}
}

func TestReplayServerUpdateReplayAllowsDataplaneKubernetesChangesForDedicatedReplay(t *testing.T) {
	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")

	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	replay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "replay-1",
			Namespace: namespace,
		},
		Spec: cloudv1.ReplaySpec{
			State:             cloudv1.ReplayStateDeploy,
			TargetClusterType: cloudv1.ReplayClusterTypeDedicated,
		},
	}
	kubeClient := fake.NewClientBuilder().WithScheme(scheme.Scheme).WithObjects(replay).Build()
	server := NewReplayServer(kubeClient)

	_, err := server.UpdateReplay(t.Context(), &apiv1.UpdateReplayRequest{
		Replay: &apiv1.Replay{
			ReplayId:      "replay-1",
			ProjectId:     "project-1",
			Provider:      "ack",
			State:         apiv1.Replay_STATE_DEPLOY,
			SourceCluster: dedicatedReplayClusterRef("source-1", "project-1", consts.ProviderTypeACK, "ap-southeast-1"),
			TargetCluster: dedicatedReplayClusterRef("target-1", "project-1", consts.ProviderTypeACK, "ap-southeast-1"),
			Tasks: []*apiv1.Replay_ReplayTask{
				{
					Name: replayTaskGroup0Name,
					Source: &apiv1.Replay_DataSource{
						AuditLogs: []*apiv1.Replay_AuditLogAddr{
							{
								Source: &apiv1.Replay_AuditLogAddr_Aliyun{
									Aliyun: &apiv1.Replay_AliyunAuditLog{
										BucketUrl: "s3://bucket/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
									},
								},
							},
						},
					},
					Target: &apiv1.Replay_DataTarget{
						Endpoint: "tidb.example.com:4000",
					},
				},
			},
			DataplaneKubernetes: &apiv1.Kubernetes{
				Name:      replayNewDataplaneName,
				Endpoint:  "https://new.example.com:6443",
				Ca:        "new-ca",
				AccountId: "654321",
				Vpc:       "vpc-new",
				Version:   "1.30",
				Info: &apiv1.Kubernetes_Alicloud_{
					Alicloud: &apiv1.Kubernetes_Alicloud{
						Vswitchs: []string{"vsw-new"},
					},
				},
			},
		},
	})
	if err != nil {
		t.Fatalf("UpdateReplay() error = %v", err)
	}

	updatedReplay := &cloudv1.Replay{}
	if err := kubeClient.Get(t.Context(), client.ObjectKey{Name: "replay-1", Namespace: namespace}, updatedReplay); err != nil {
		t.Fatalf("Get(replay) error = %v", err)
	}
	if got := updatedReplay.Spec.Kubernetes.Name; got != replayNewDataplaneName {
		t.Fatalf("updated replay kubernetes.name = %q, want %q", got, replayNewDataplaneName)
	}
}

func TestReplayServerUpdateReplayAllowsPartialAWSSubnetInput(t *testing.T) {
	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")

	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	replay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "replay-1",
			Namespace: namespace,
		},
		Spec: cloudv1.ReplaySpec{
			State:             cloudv1.ReplayStateDeploy,
			TargetClusterType: cloudv1.ReplayClusterTypeDedicated,
			Kubernetes: cloudv1.Kubernetes{
				Cloud: cloudv1.Cloud{
					AWS: &cloudv1.CloudAWS{},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().WithScheme(scheme.Scheme).WithObjects(replay).Build()
	server := NewReplayServer(kubeClient)

	_, err := server.UpdateReplay(t.Context(), &apiv1.UpdateReplayRequest{
		Replay: &apiv1.Replay{
			ReplayId:      "replay-1",
			ProjectId:     "project-1",
			Provider:      consts.ProviderTypeEKS,
			State:         apiv1.Replay_STATE_DEPLOY,
			SourceCluster: dedicatedReplayClusterRef("source-1", "project-1", consts.ProviderTypeEKS, "us-west-2"),
			TargetCluster: dedicatedReplayClusterRef("target-1", "project-1", consts.ProviderTypeEKS, "us-west-2"),
			Tasks: []*apiv1.Replay_ReplayTask{
				{
					Name: replayTaskGroup0Name,
					Source: &apiv1.Replay_DataSource{
						AuditLogs: []*apiv1.Replay_AuditLogAddr{
							{
								Source: &apiv1.Replay_AuditLogAddr_Aws{
									Aws: &apiv1.Replay_AWSAuditLog{
										S3Url:   "s3://bucket/prefix",
										RoleArn: "arn:aws:iam::123456789012:role/replay",
									},
								},
							},
						},
					},
					Target: &apiv1.Replay_DataTarget{
						Endpoint: "tidb.example.com:4000",
					},
				},
			},
			DataplaneKubernetes: &apiv1.Kubernetes{
				Info: &apiv1.Kubernetes_Aws_{
					Aws: &apiv1.Kubernetes_Aws{
						NodeSubnets: []*apiv1.Kubernetes_AwsSubnet{
							{Id: "subnet-new-node", AzName: "us-west-2a"},
						},
					},
				},
			},
		},
	})
	if err != nil {
		t.Fatalf("UpdateReplay() error = %v", err)
	}

	updatedReplay := &cloudv1.Replay{}
	if err := kubeClient.Get(t.Context(), client.ObjectKey{Name: "replay-1", Namespace: namespace}, updatedReplay); err != nil {
		t.Fatalf("Get(replay) error = %v", err)
	}
	aws := updatedReplay.Spec.Kubernetes.Cloud.AWS
	if aws == nil {
		t.Fatal("updated replay cloud.aws = nil, want partial aws config")
	}
	if len(aws.SubnetGroups) != 1 {
		t.Fatalf("updated replay aws subnet groups = %+v, want node subnet group only", aws.SubnetGroups)
	}
	if aws.SubnetGroups[0].Name != cloudv1.AWSSubnetGroupNodes ||
		!reflect.DeepEqual(aws.SubnetGroups[0].Subnets, []cloudv1.AWSSubnet{{SubnetID: "subnet-new-node", Zone: "us-west-2a"}}) {
		t.Fatalf("updated replay node subnet group = %+v, want request node subnet only", aws.SubnetGroups[0])
	}
}

func TestReplayServerUpdateReplayPremiumDataplaneOnlyUpdateReturnsInternalBuildError(t *testing.T) {
	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")

	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	replay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "replay-1",
			Namespace: namespace,
		},
		Spec: cloudv1.ReplaySpec{
			TargetClusterID:   "target-premium",
			State:             cloudv1.ReplayStateDeploy,
			TargetClusterType: cloudv1.ReplayClusterTypePremium,
			TargetProvider:    "ack",
			TargetRegion:      "ap-southeast-1",
			SourceClusterID:   ptr.To("source-premium"),
			SourceClusterType: ptr.To(cloudv1.ReplayClusterTypePremium),
			SourceProvider:    ptr.To("ack"),
			SourceRegion:      ptr.To("ap-southeast-1"),
			Tasks: []cloudv1.ReplayTask{
				{
					Name:           replayTaskGroup0Name,
					TargetEndpoint: "tidb.example.com:4000",
					SourceAuditLogAddrs: []cloudv1.AuditLogAddr{
						{
							Aliyun: &cloudv1.AliyunAuditLog{
								BucketURL: "s3://bucket/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
							},
						},
					},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().WithScheme(scheme.Scheme).WithObjects(replay).Build()
	server := NewReplayServer(kubeClient)

	originalReplay := replay.DeepCopy()
	err := assertUpdateReplayRejectedAndSpecUnchanged(t, server, kubeClient, namespace, &apiv1.Replay{
		ReplayId:  "replay-1",
		ProjectId: "project-1",
		Provider:  "ack",
		State:     apiv1.Replay_STATE_DEPLOY,
		DataplaneKubernetes: &apiv1.Kubernetes{
			Name: replayNewDataplaneName,
			Info: &apiv1.Kubernetes_Alicloud_{
				Alicloud: &apiv1.Kubernetes_Alicloud{
					Vswitchs: []string{"vsw-new"},
				},
			},
		},
	}, originalReplay)
	if status.Code(err) != codes.Internal {
		t.Fatalf("UpdateReplay() code = %v, want %v", status.Code(err), codes.Internal)
	}
}

func TestReplayBuilderBuildForUpdateDedicatedSourcePremiumTargetRebuildsRequestTaskMaterial(t *testing.T) {
	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "p1")
	sourceCluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      replayOldSourceClusterID,
			Namespace: namespace,
			Labels: map[string]string{
				cloudv1.ProjectTagKey:     "p1",
				cloudv1.ProviderTypeLabel: consts.ProviderTypeEKS,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Audit: &cloudv1.Audit{
						Enable:    true,
						BucketURL: "s3://cluster-audit/prefix",
						RoleArn:   "arn:aws:iam::123456789012:role/replay",
					},
				},
			},
		},
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		WithObjects(sourceCluster).
		Build()

	builder := newReplayBuilder(NewReplayServer(kubeClient), &apiv1.Replay{
		ReplayId:              "r1",
		ProjectId:             "p1",
		Provider:              "ack",
		Region:                "ap-southeast-1",
		PremiumTargetEndpoint: ptr.To("tidb.request.example.com:4000"),
		SourceCluster:         dedicatedReplayClusterRef(replayOldSourceClusterID, "p1", consts.ProviderTypeEKS, "us-west-2"),
		TargetCluster:         premiumReplayClusterRef("target-premium", "p1", consts.ProviderTypeACK, "ap-southeast-1"),
		DataplaneKubernetes: &apiv1.Kubernetes{
			Name:     replayNewDataplaneName,
			Endpoint: "https://dp-request.example.com:6443",
		},
		Defaults: &apiv1.Replay_ReplayDefaults{
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aws{
					Aws: &apiv1.Replay_AWSAuditLog{
						S3Url: "s3://request-audit/prefix",
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			State:             cloudv1.ReplayStateDeploy,
			TargetClusterType: cloudv1.ReplayClusterTypePremium,
			SourceClusterID:   ptr.To(replayOldSourceClusterID),
			SourceClusterType: ptr.To(cloudv1.ReplayClusterTypeDedicated),
			SourceProvider:    ptr.To(consts.ProviderTypeEKS),
			SourceRegion:      ptr.To("us-west-2"),
			Tasks: []cloudv1.ReplayTask{
				{
					Name:           "group-old",
					TargetEndpoint: "tidb.old.example.com:4000",
					SourceAuditLogAddrs: []cloudv1.AuditLogAddr{
						{
							AWS: &cloudv1.AWSAuditLog{
								S3URL: "s3://old-audit/prefix",
							},
						},
					},
				},
			},
		},
	}

	if err := builder.BuildForUpdate(t.Context(), replay); err != nil {
		t.Fatalf("BuildForUpdate() error = %v", err)
	}

	if len(replay.Spec.Tasks) != 1 {
		t.Fatalf("len(tasks) = %d, want 1", len(replay.Spec.Tasks))
	}
	task := replay.Spec.Tasks[0]
	if task.Name != replayTaskGroupDefaultName {
		t.Fatalf("Name = %q, want rebuilt premium task name group-default", task.Name)
	}
	if task.TargetEndpoint != "tidb.request.example.com:4000" {
		t.Fatalf("TargetEndpoint = %q, want request endpoint", task.TargetEndpoint)
	}
	if len(task.SourceAuditLogAddrs) != 1 ||
		task.SourceAuditLogAddrs[0].AWS == nil ||
		task.SourceAuditLogAddrs[0].AWS.S3URL != "s3://request-audit/prefix" {
		t.Fatalf("SourceAuditLogAddrs = %#v, want request audit log only", task.SourceAuditLogAddrs)
	}
	if got := ptr.Deref(replay.Spec.SourceClusterID, ""); got != replayOldSourceClusterID {
		t.Fatalf("SourceClusterID = %q, want %q", got, replayOldSourceClusterID)
	}
	if replay.Spec.Kubernetes.Name != replayNewDataplaneName ||
		replay.Spec.Kubernetes.Endpoint != "https://dp-request.example.com:6443" {
		t.Fatalf("Kubernetes = %#v, want explicit dataplane metadata", replay.Spec.Kubernetes)
	}
}

func TestReplayBuilderBuildForUpdateDedicatedSourcePremiumTargetRejectsPartialTaskMaterialUpdate(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		Provider:  "ack",
		Region:    "ap-southeast-1",
		TargetCluster: premiumReplayClusterRef(
			"target-premium",
			"p1",
			consts.ProviderTypeACK,
			"ap-southeast-1",
		),
		Defaults: &apiv1.Replay_ReplayDefaults{
			Replicas: ptr.To(int32(3)),
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			State:             cloudv1.ReplayStateDeploy,
			TargetClusterType: cloudv1.ReplayClusterTypePremium,
			SourceClusterID:   ptr.To(replayOldSourceClusterID),
			SourceClusterType: ptr.To(cloudv1.ReplayClusterTypeDedicated),
			SourceProvider:    ptr.To("ack"),
			SourceRegion:      ptr.To("ap-southeast-1"),
			Tasks: []cloudv1.ReplayTask{
				{
					Name:           "group-old",
					TargetEndpoint: "tidb.old.example.com:4000",
					Replicas:       1,
					SourceAuditLogAddrs: []cloudv1.AuditLogAddr{
						{
							Aliyun: &cloudv1.AliyunAuditLog{
								BucketURL: "s3://old-audit/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
							},
						},
					},
				},
			},
		},
	}

	err := assertBuildForUpdateRejected(t, builder, replay)
	if !replayErrorMatches(err, errReplaySourceClusterMissing) {
		t.Fatalf("BuildForUpdate() error = %v, want %v", err, errReplaySourceClusterMissing)
	}
}

func TestReplayBuilderBuildForUpdatePremiumMultipleExistingTasksCanBeFullyRebuilt(t *testing.T) {
	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:              "r1",
		ProjectId:             "p1",
		Provider:              "ack",
		Region:                "ap-southeast-1",
		PremiumTargetEndpoint: ptr.To("tidb.request.example.com:4000"),
		SourceCluster:         premiumReplayClusterRef(replayOldSourceClusterID, "p1", consts.ProviderTypeACK, "ap-southeast-1"),
		TargetCluster:         premiumReplayClusterRef("target-premium", "p1", consts.ProviderTypeACK, "ap-southeast-1"),
		DataplaneKubernetes: &apiv1.Kubernetes{
			Name:     replayNewDataplaneName,
			Endpoint: "https://dp-request.example.com:6443",
		},
		Defaults: &apiv1.Replay_ReplayDefaults{
			Replicas: ptr.To(int32(3)),
			SourceAuditLog: &apiv1.Replay_AuditLogAddr{
				Source: &apiv1.Replay_AuditLogAddr_Aliyun{
					Aliyun: &apiv1.Replay_AliyunAuditLog{
						BucketUrl: "s3://request-audit/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com",
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			State:             cloudv1.ReplayStateDeploy,
			TargetClusterType: cloudv1.ReplayClusterTypePremium,
			SourceClusterID:   ptr.To(replayOldSourceClusterID),
			SourceClusterType: ptr.To(cloudv1.ReplayClusterTypePremium),
			SourceProvider:    ptr.To("ack"),
			SourceRegion:      ptr.To("ap-southeast-1"),
			Tasks: []cloudv1.ReplayTask{
				{Name: replayTaskGroup0Name, TargetEndpoint: "tidb.old.example.com:4000"},
				{Name: "group-1", TargetEndpoint: "tidb.old.example.com:4000"},
			},
		},
	}

	if err := builder.BuildForUpdate(t.Context(), replay); err != nil {
		t.Fatalf("BuildForUpdate() error = %v", err)
	}
	if len(replay.Spec.Tasks) != 1 {
		t.Fatalf("len(tasks) = %d, want 1", len(replay.Spec.Tasks))
	}
	task := replay.Spec.Tasks[0]
	if task.Name != replayTaskGroup0Name {
		t.Fatalf("Name = %q, want rebuilt premium task", task.Name)
	}
	if task.TargetEndpoint != "tidb.request.example.com:4000" {
		t.Fatalf("TargetEndpoint = %q, want request endpoint", task.TargetEndpoint)
	}
	if task.Replicas != 3 {
		t.Fatalf("Replicas = %d, want 3", task.Replicas)
	}
	if len(task.SourceAuditLogAddrs) != 1 ||
		task.SourceAuditLogAddrs[0].Aliyun == nil ||
		task.SourceAuditLogAddrs[0].Aliyun.BucketURL != "s3://request-audit/prefix?endpoint=http://oss-cn-hangzhou.aliyuncs.com" {
		t.Fatalf("SourceAuditLogAddrs = %#v, want request audit log", task.SourceAuditLogAddrs)
	}
	if got := ptr.Deref(replay.Spec.SourceClusterID, ""); got != replayOldSourceClusterID {
		t.Fatalf("SourceClusterID = %q, want %q", got, replayOldSourceClusterID)
	}
	if replay.Spec.Kubernetes.Name != replayNewDataplaneName ||
		replay.Spec.Kubernetes.Endpoint != "https://dp-request.example.com:6443" {
		t.Fatalf("Kubernetes = %#v, want explicit dataplane metadata", replay.Spec.Kubernetes)
	}
}

func TestReplayBuilderBuildForUpdatePremiumShapeChangesStillRequireCompleteRebuildMaterial(t *testing.T) {
	const targetOldClusterID = "target-old"

	tests := []struct {
		name         string
		buildReq     func() *apiv1.Replay
		buildReplay  func() *cloudv1.Replay
		mutateReq    func(*apiv1.Replay)
		mutateReplay func(*cloudv1.Replay)
	}{
		{
			name:        "source cluster change",
			buildReq:    newPremiumBuildForUpdateRequest,
			buildReplay: newPremiumBuildForUpdateReplay,
			mutateReq: func(req *apiv1.Replay) {
				req.SourceCluster.ClusterId = "source-new"
			},
			mutateReplay: func(replay *cloudv1.Replay) {
				replay.Spec.SourceClusterID = ptr.To(replayOldSourceClusterID)
			},
		},
		{
			name:        "target cluster change",
			buildReq:    newPremiumBuildForUpdateRequest,
			buildReplay: newPremiumBuildForUpdateReplay,
			mutateReq: func(req *apiv1.Replay) {
				req.TargetCluster.ClusterId = "target-new"
			},
			mutateReplay: func(replay *cloudv1.Replay) {
				replay.Spec.TargetClusterID = targetOldClusterID
			},
		},
		{
			name: "target type override",
			buildReq: func() *apiv1.Replay {
				return &apiv1.Replay{
					ReplayId:  "r1",
					ProjectId: "p1",
					Provider:  consts.ProviderTypeACK,
					Region:    "ap-southeast-1",
					TargetCluster: &apiv1.Replay_ClusterRef{
						ClusterId:   "target-premium",
						ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
						Provider:    consts.ProviderTypeACK,
						Region:      "ap-southeast-1",
					},
				}
			},
			buildReplay: func() *cloudv1.Replay {
				return &cloudv1.Replay{
					Spec: cloudv1.ReplaySpec{
						State:             cloudv1.ReplayStateDeploy,
						TargetClusterType: cloudv1.ReplayClusterTypePremium,
						TargetClusterID:   "target-premium",
						TargetProvider:    consts.ProviderTypeACK,
						TargetRegion:      "ap-southeast-1",
					},
				}
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			req := tt.buildReq()
			if tt.mutateReq != nil {
				tt.mutateReq(req)
			}
			replay := tt.buildReplay()
			if tt.mutateReplay != nil {
				tt.mutateReplay(replay)
			}

			builder := newReplayBuilder(nil, req)
			if err := assertBuildForUpdateRejected(t, builder, replay); err == nil {
				t.Fatal("assertBuildForUpdateRejected() error = nil, want rejected update")
			}
		})
	}
}

func TestReplayServerUpdateReplayPremiumShapeChangesReturnInternalBuildError(t *testing.T) {
	namespace := genNamespace(config2.GetConfig().NamespacePrefix, "project-1")
	const targetOldClusterID = "target-old"

	if err := cloudv1.AddToScheme(scheme.Scheme); err != nil {
		t.Fatalf("AddToScheme() error = %v", err)
	}

	tests := []struct {
		name         string
		mutateReq    func(*apiv1.Replay)
		mutateReplay func(*cloudv1.Replay)
	}{
		{
			name: "source cluster change",
			mutateReq: func(req *apiv1.Replay) {
				req.SourceCluster.ClusterId = "source-new"
			},
			mutateReplay: func(replay *cloudv1.Replay) {
				replay.Spec.SourceClusterID = ptr.To(replayOldSourceClusterID)
			},
		},
		{
			name: "target cluster change",
			mutateReq: func(req *apiv1.Replay) {
				req.TargetCluster.ClusterId = "target-new"
			},
			mutateReplay: func(replay *cloudv1.Replay) {
				replay.Spec.TargetClusterID = targetOldClusterID
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			replay := newPremiumUpdateReplay(namespace)
			if tt.mutateReplay != nil {
				tt.mutateReplay(replay)
			}
			kubeClient := fake.NewClientBuilder().WithScheme(scheme.Scheme).WithObjects(replay).Build()
			server := NewReplayServer(kubeClient)

			req := newPremiumUpdateReplayRequest()
			if tt.mutateReq != nil {
				tt.mutateReq(req)
			}

			originalReplay := replay.DeepCopy()
			err := assertUpdateReplayRejectedAndSpecUnchanged(t, server, kubeClient, namespace, req, originalReplay)
			if status.Code(err) != codes.Internal {
				t.Fatalf("UpdateReplay() code = %v, want %v", status.Code(err), codes.Internal)
			}
			if !strings.Contains(err.Error(), "failed to get premium cluster") {
				t.Fatalf("UpdateReplay() error = %v, want premium cluster lookup failure", err)
			}
		})
	}
}

func TestReplayBuilderSetReplayConfigForUpdateDoesNotMutateClusterInfo(t *testing.T) {
	const targetOldClusterID = "target-old"

	builder := newReplayBuilder(nil, &apiv1.Replay{
		ReplayId:  "r1",
		ProjectId: "p1",
		Provider:  "ack",
		Region:    "ap-southeast-1",
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "target-new",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_PREMIUM,
			Provider:    "ack",
			Region:      "ap-southeast-1",
		},
		SourceCluster: &apiv1.Replay_ClusterRef{
			ClusterId:   "source-new",
			ClusterType: apiv1.Replay_CLUSTER_TYPE_DEDICATED,
			Provider:    "ack",
			Region:      "ap-southeast-1",
		},
		Tasks: []*apiv1.Replay_ReplayTask{
			{
				Name: replayTaskGroup0Name,
				Target: &apiv1.Replay_DataTarget{
					Endpoint: "tidb.example.com:4000",
				},
				Source: &apiv1.Replay_DataSource{
					AuditLogs: []*apiv1.Replay_AuditLogAddr{
						{
							Source: &apiv1.Replay_AuditLogAddr_Aliyun{
								Aliyun: &apiv1.Replay_AliyunAuditLog{
									BucketUrl: "s3://bucket/prefix",
								},
							},
						},
					},
				},
			},
		},
	})

	replay := &cloudv1.Replay{
		Spec: cloudv1.ReplaySpec{
			TargetClusterID:   targetOldClusterID,
			TargetClusterType: cloudv1.ReplayClusterTypeDedicated,
			TargetProvider:    "eks",
			TargetRegion:      "us-west-2",
			SourceClusterID:   ptr.To(replayOldSourceClusterID),
			SourceClusterType: ptr.To(cloudv1.ReplayClusterTypePremium),
			SourceProvider:    ptr.To("eks"),
			SourceRegion:      ptr.To("us-west-2"),
		},
	}

	builder.setReplayConfig(replay)

	if replay.Spec.TargetClusterID != targetOldClusterID {
		t.Fatalf("TargetClusterID = %q, want %q", replay.Spec.TargetClusterID, targetOldClusterID)
	}
	if replay.Spec.TargetClusterType != cloudv1.ReplayClusterTypeDedicated {
		t.Fatalf("TargetClusterType = %q, want %q", replay.Spec.TargetClusterType, cloudv1.ReplayClusterTypeDedicated)
	}
	if replay.Spec.TargetProvider != "eks" || replay.Spec.TargetRegion != "us-west-2" {
		t.Fatalf("target provider/region = %q/%q, want eks/us-west-2", replay.Spec.TargetProvider, replay.Spec.TargetRegion)
	}
	if replay.Spec.SourceClusterID == nil || *replay.Spec.SourceClusterID != replayOldSourceClusterID {
		t.Fatalf("SourceClusterID = %v, want %s", replay.Spec.SourceClusterID, replayOldSourceClusterID)
	}
	if replay.Spec.SourceClusterType == nil || *replay.Spec.SourceClusterType != cloudv1.ReplayClusterTypePremium {
		t.Fatalf("SourceClusterType = %v, want %q", replay.Spec.SourceClusterType, cloudv1.ReplayClusterTypePremium)
	}
	if replay.Spec.SourceProvider == nil || *replay.Spec.SourceProvider != "eks" {
		t.Fatalf("SourceProvider = %v, want eks", replay.Spec.SourceProvider)
	}
	if replay.Spec.SourceRegion == nil || *replay.Spec.SourceRegion != "us-west-2" {
		t.Fatalf("SourceRegion = %v, want us-west-2", replay.Spec.SourceRegion)
	}
}