package service

import (
	"context"
	"errors"
	"net/http"
	"testing"
	"time"

	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/pkgv2/clusterproxy/tidbcloud"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade/mock_facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo/mock_repo"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/mock/gomock"
	corev1 "k8s.io/api/core/v1"
)

func setConfigForTest(t *testing.T, cfg *config.CentralConfig) {
	t.Helper()

	old := *config.Global()
	config.Set(cfg)
	t.Cleanup(func() {
		config.Set(&old)
	})
}

func TestResolveNewArchShardUsesFixedShareRulePriority(t *testing.T) {
	setConfigForTest(t, &config.CentralConfig{
		CDCConfig: config.CDCConfig{NewArch: config.CDCNewArchConfig{
			ShareScope: "global",
			ShareRules: []config.CDCNewArchShareRule{
				{ShareScope: "tenant", TenantIDs: []string{"tenant-1"}},
				{ShareScope: "cluster", ClusterIDs: []string{"cluster-1"}},
				{ShareScope: "changefeed", ClusterIDs: []string{"cluster-1"}},
			},
		}},
	})
	changefeed := &entity.CDCChangefeed{
		ID:            "cf-1",
		TenantID:      "tenant-1",
		ClusterID:     "cluster-1",
		CloudProvider: "aws",
		Region:        "us-west-2",
		Kubernetes:    "k8s-1",
	}

	shard := resolveNewArchShard(changefeed)

	require.Equal(t, "changefeed", shard.ShareScope)
	require.Equal(t, "cf-1", shard.ShareKey)
	require.NotEmpty(t, shard.ClusterName)
}

func TestConvertServerlessChangefeedConfigToV2PreservesColumnSelectors(t *testing.T) {
	input := &serverless.ChangefeedConfig{
		ReplicaConfig: &serverless.ReplicaConfig{
			Sink: &serverless.SinkConfig{
				ColumnSelectors: []*serverless.ColumnSelector{{
					Matcher: []string{"app.orders"},
					Columns: []string{"id", "status"},
				}},
			},
		},
	}

	converted, err := convertServerlessChangefeedConfigToV2(input)
	require.NoError(t, err)
	require.Len(t, converted.ReplicaConfig.Sink.ColumnSelectors, 1)
	require.Equal(t, []string{"app.orders"}, converted.ReplicaConfig.Sink.ColumnSelectors[0].Matcher)
	require.Equal(t, []string{"id", "status"}, converted.ReplicaConfig.Sink.ColumnSelectors[0].Columns)
}

func TestConvertServerlessChangefeedConfigToV2PreservesRecoveryFields(t *testing.T) {
	bdrMode := true
	firstTS := uint64(123)
	secondTS := uint64(456)
	input := &serverless.ChangefeedConfig{
		ReplicaConfig: &serverless.ReplicaConfig{
			BDRMode: &bdrMode,
			Filter: &serverless.FilterConfig{
				IgnoreTxnStartTS: []*uint64{&firstTS, &secondTS},
			},
		},
	}

	converted, err := convertServerlessChangefeedConfigToV2(input)
	require.NoError(t, err)
	require.NotNil(t, converted.ReplicaConfig.BDRMode)
	require.True(t, *converted.ReplicaConfig.BDRMode)
	require.Equal(t, []uint64{123, 456}, converted.ReplicaConfig.Filter.IgnoreTxnStartTs)
}

func TestResolveNewArchShardCanSelectChangefeedScopeByClusterRule(t *testing.T) {
	setConfigForTest(t, &config.CentralConfig{
		CDCConfig: config.CDCConfig{NewArch: config.CDCNewArchConfig{
			ShareScope: "global",
			ShareRules: []config.CDCNewArchShareRule{{
				ShareScope: "changefeed",
				ClusterIDs: []string{"cluster-1"},
			}},
		}},
	})
	changefeed := &entity.CDCChangefeed{
		ID:            "cf-1",
		TenantID:      "tenant-1",
		ClusterID:     "cluster-1",
		CloudProvider: "aws",
		Region:        "us-west-2",
		Kubernetes:    "k8s-1",
	}

	shard := resolveNewArchShard(changefeed)

	require.Equal(t, "changefeed", shard.ShareScope)
	require.Equal(t, "cf-1", shard.ShareKey)
}

func TestResolveNewArchShardFallsBackToDefaultShareScope(t *testing.T) {
	setConfigForTest(t, &config.CentralConfig{
		CDCConfig: config.CDCConfig{NewArch: config.CDCNewArchConfig{ShareScope: "cluster"}},
	})
	changefeed := &entity.CDCChangefeed{
		ID:            "cf-1",
		TenantID:      "tenant-1",
		ClusterID:     "cluster-1",
		CloudProvider: "aws",
		Region:        "us-west-2",
		Kubernetes:    "k8s-1",
	}

	shard := resolveNewArchShard(changefeed)

	require.Equal(t, "cluster", shard.ShareScope)
	require.Equal(t, "cluster-1", shard.ShareKey)
}

func TestEnsureNewArchChangefeedResourceKeepsExistingShard(t *testing.T) {
	setConfigForTest(t, &config.CentralConfig{
		CDCConfig: config.CDCConfig{NewArch: config.CDCNewArchConfig{ShareScope: "changefeed"}},
	})
	existing := &entity.TiCDCChangefeedResource{
		ClusterID:             "cluster-1",
		ChangefeedID:          "cf-1",
		CloudProvider:         "aws",
		Region:                "us-west-2",
		Kubernetes:            "k8s-1",
		ShareScope:            "tenant",
		ShareKey:              "tenant-1",
		TiCDCClusterName:      "tenant-shard",
		ExpectedSpecification: "2rcu",
	}
	repo := &fakeTiCDCChangefeedResourceRepo{resource: existing}
	svc := &cdcService{tiCDCChangefeedResourceRepo: repo}

	shard, err := svc.ensureNewArchChangefeedResource(t.Context(), &entity.CDCChangefeed{
		ID:            "cf-1",
		TenantID:      "tenant-1",
		ClusterID:     "cluster-1",
		CloudProvider: "aws",
		Region:        "us-west-2",
		Kubernetes:    "k8s-1",
	}, "4rcu")

	require.Nil(t, err)
	require.Equal(t, "tenant", shard.ShareScope)
	require.Equal(t, "tenant-1", shard.ShareKey)
	require.Equal(t, "tenant-shard", shard.ClusterName)
	require.Equal(t, "tenant", repo.resource.ShareScope)
	require.Equal(t, "tenant-1", repo.resource.ShareKey)
	require.Equal(t, "tenant-shard", repo.resource.TiCDCClusterName)
	require.Equal(t, "4rcu", repo.resource.ExpectedSpecification)
}

func TestIsNewArchChangefeedResourceReleasedTreatsMissingResourceAsReleased(t *testing.T) {
	svc := &cdcService{tiCDCChangefeedResourceRepo: &fakeTiCDCChangefeedResourceRepo{}}

	released, err := svc.isNewArchChangefeedResourceReleased(t.Context(), "cluster-1", "cf-1")

	require.Nil(t, err)
	require.True(t, released)
}

func TestEnsureNewArchTiCDCChangefeedRecordReusesDeletedRecord(t *testing.T) {
	deletedAt := time.Now().UTC().Add(-time.Minute)
	repo := &fakeKernelTiCDCChangefeedRepo{
		record: &entity.KernelTiCDCChangefeed{
			ID:                 12,
			ClusterID:          "cluster-1",
			ChangefeedID:       "cf-1",
			KernelChangefeedID: "cf-1",
			KernelStatus:       constant.TiCDCChangefeedKernelStatusDeleted,
			DeletedAt:          &deletedAt,
		},
	}
	svc := &cdcService{tiCDCChangefeedRepo: repo}

	err := svc.ensureNewArchTiCDCChangefeedRecord(
		t.Context(),
		&entity.CDCChangefeed{
			ID:         "cf-1",
			ClusterID:  "cluster-1",
			KeySpaceID: 1290905,
		},
		&newArchShard{
			Provider:    "aws",
			Region:      "us-east-1",
			Kubernetes:  "staging-us-east-1-f01",
			ShareScope:  "cluster",
			ShareKey:    "cluster-1",
			ClusterName: "ess-ticdc-us-east-1-test",
		},
		constant.TiCDCChangefeedKernelStatusDeleted,
	)

	require.Nil(t, err)
	require.Equal(t, uint64(12), repo.record.ID)
	require.Equal(t, constant.TiCDCChangefeedKernelStatusDeleted, repo.record.KernelStatus)
	require.NotNil(t, repo.record.DeletedAt)
}

func TestPersistNewArchTiCDCChangefeedCreateIntentReturnsExistingShard(t *testing.T) {
	setConfigForTest(t, &config.CentralConfig{
		CDCConfig: config.CDCConfig{NewArch: config.CDCNewArchConfig{ShareScope: "changefeed"}},
	})
	existing := &entity.KernelTiCDCChangefeed{
		ClusterID:          "cluster-1",
		ChangefeedID:       "cf-1",
		CloudProvider:      "aws",
		Region:             "us-west-2",
		Kubernetes:         "k8s-1",
		ShareScope:         "tenant",
		ShareKey:           "tenant-1",
		TiCDCClusterName:   "tenant-shard",
		KernelChangefeedID: "cf-1",
		KernelStatus:       constant.TiCDCChangefeedKernelStatusPending,
	}
	repo := &fakeKernelTiCDCChangefeedRepo{record: existing}
	svc := &cdcService{tiCDCChangefeedRepo: repo}

	configShard := resolveNewArchShard(&entity.CDCChangefeed{
		ID:            "cf-1",
		TenantID:      "tenant-1",
		ClusterID:     "cluster-1",
		CloudProvider: "aws",
		Region:        "us-west-2",
		Kubernetes:    "k8s-1",
		KeySpaceID:    uint32(123),
	})
	record, shard, err := svc.persistNewArchTiCDCChangefeedCreateIntent(t.Context(), &entity.CDCChangefeed{
		ID:            "cf-1",
		TenantID:      "tenant-1",
		ClusterID:     "cluster-1",
		CloudProvider: "aws",
		Region:        "us-west-2",
		Kubernetes:    "k8s-1",
		KeySpaceID:    uint32(123),
	}, &configShard)

	require.Nil(t, err)
	require.Equal(t, record, repo.record)
	require.Equal(t, "tenant", shard.ShareScope)
	require.Equal(t, "tenant-1", shard.ShareKey)
	require.Equal(t, "tenant-shard", shard.ClusterName)
	require.Equal(t, "tenant", repo.record.ShareScope)
	require.Equal(t, "tenant-1", repo.record.ShareKey)
	require.Equal(t, "tenant-shard", repo.record.TiCDCClusterName)
	require.Equal(t, constant.TiCDCChangefeedKernelStatusPending, repo.record.KernelStatus)
}

// TestSelectNewArchTiCDCInstance pins down the picker contract:
//
//   - sub-smallest demand snaps up to the smallest instance, 1 replica
//   - between adjacent instance types, snap to whichever side of the
//     midpoint the demand falls on
//   - exceeding the largest type's single-replica capacity stays on the
//     largest type and scales by replica count (round half-up)
//
// The single-instance-type default (the shipped default of a 16c
// catalog) is the most common production shape — even a 2rcu shard
// materializes a 16c × 1 cluster — so it gets its own case.
func TestSelectNewArchTiCDCInstance(t *testing.T) {
	multiCatalog := []config.CDCNewArchInstanceType{
		{Name: "small", CpuPerReplica: 2, MemoryGi: 8, StorageGi: 50, Iops: 4000, ThroughputMbs: 400},
		{Name: "medium", CpuPerReplica: 4, MemoryGi: 16, StorageGi: 50, Iops: 4000, ThroughputMbs: 400},
		{Name: "large", CpuPerReplica: 8, MemoryGi: 32, StorageGi: 100, Iops: 6000, ThroughputMbs: 500},
		{Name: "xlarge", CpuPerReplica: 16, MemoryGi: 64, StorageGi: 200, Iops: 12000, ThroughputMbs: 1000},
	}
	singleCatalog := []config.CDCNewArchInstanceType{
		{Name: "16c", CpuPerReplica: 16, MemoryGi: 64, StorageGi: 200, Iops: 12000, ThroughputMbs: 1000},
	}

	tests := []struct {
		name             string
		catalog          []config.CDCNewArchInstanceType
		cpuPerRcu        uint32
		rcu              uint32
		wantInstanceName string
		wantReplicas     uint32
	}{
		// Multi-tier catalog: behaviour mirrors the legacy
		// specification ladder semantics.
		{
			name:             "multi: rcu=0 maps via fallback=3 -> medium 4cpu",
			catalog:          multiCatalog,
			cpuPerRcu:        1,
			rcu:              0,
			wantInstanceName: "medium",
			wantReplicas:     1,
		},
		{
			name:             "multi: under smallest snaps up",
			catalog:          multiCatalog,
			cpuPerRcu:        1,
			rcu:              1,
			wantInstanceName: "small",
			wantReplicas:     1,
		},
		{
			name:             "multi: exact smallest",
			catalog:          multiCatalog,
			cpuPerRcu:        1,
			rcu:              2,
			wantInstanceName: "small",
			wantReplicas:     1,
		},
		{
			name:             "multi: 3 below mid(small,medium)=3 -> small",
			catalog:          multiCatalog,
			cpuPerRcu:        1,
			rcu:              3,
			wantInstanceName: "medium",
			wantReplicas:     1,
		},
		{
			name:             "multi: 4 exact medium",
			catalog:          multiCatalog,
			cpuPerRcu:        1,
			rcu:              4,
			wantInstanceName: "medium",
			wantReplicas:     1,
		},
		{
			name:             "multi: 5 above mid(medium,large)=6 -> medium",
			catalog:          multiCatalog,
			cpuPerRcu:        1,
			rcu:              5,
			wantInstanceName: "medium",
			wantReplicas:     1,
		},
		{
			name:             "multi: 7 above mid -> large",
			catalog:          multiCatalog,
			cpuPerRcu:        1,
			rcu:              7,
			wantInstanceName: "large",
			wantReplicas:     1,
		},
		{
			name:             "multi: 16 exact xlarge",
			catalog:          multiCatalog,
			cpuPerRcu:        1,
			rcu:              16,
			wantInstanceName: "xlarge",
			wantReplicas:     1,
		},
		{
			name:             "multi: 24 exceeds largest -> xlarge x2",
			catalog:          multiCatalog,
			cpuPerRcu:        1,
			rcu:              24,
			wantInstanceName: "xlarge",
			wantReplicas:     2,
		},
		{
			name:             "multi: 32 exceeds -> xlarge x2",
			catalog:          multiCatalog,
			cpuPerRcu:        1,
			rcu:              32,
			wantInstanceName: "xlarge",
			wantReplicas:     2,
		},
		{
			name:             "multi: 64 -> xlarge x4",
			catalog:          multiCatalog,
			cpuPerRcu:        1,
			rcu:              64,
			wantInstanceName: "xlarge",
			wantReplicas:     4,
		},

		// Default single-instance catalog: every shard materializes
		// as N replicas of 16c. Sub-16 always = 1 replica.
		{
			name:             "single 16c: 2rcu -> 16c x1",
			catalog:          singleCatalog,
			cpuPerRcu:        1,
			rcu:              2,
			wantInstanceName: "16c",
			wantReplicas:     1,
		},
		{
			name:             "single 16c: 8rcu -> 16c x1",
			catalog:          singleCatalog,
			cpuPerRcu:        1,
			rcu:              8,
			wantInstanceName: "16c",
			wantReplicas:     1,
		},
		{
			name:             "single 16c: 16rcu -> 16c x1",
			catalog:          singleCatalog,
			cpuPerRcu:        1,
			rcu:              16,
			wantInstanceName: "16c",
			wantReplicas:     1,
		},
		{
			name:             "single 16c: 24rcu -> 16c x2",
			catalog:          singleCatalog,
			cpuPerRcu:        1,
			rcu:              24,
			wantInstanceName: "16c",
			wantReplicas:     2,
		},
		{
			name:             "single 16c: 32rcu -> 16c x2",
			catalog:          singleCatalog,
			cpuPerRcu:        1,
			rcu:              32,
			wantInstanceName: "16c",
			wantReplicas:     2,
		},
		{
			name:             "single 16c: 48rcu -> 16c x3",
			catalog:          singleCatalog,
			cpuPerRcu:        1,
			rcu:              48,
			wantInstanceName: "16c",
			wantReplicas:     3,
		},
		{
			name:             "single 16c: 64rcu -> 16c x4",
			catalog:          singleCatalog,
			cpuPerRcu:        1,
			rcu:              64,
			wantInstanceName: "16c",
			wantReplicas:     4,
		},

		// cpuPerRcu != 1 — verifies the conversion factor is applied.
		// With cpuPerRcu=2, every rcu doubles in cpu cost.
		{
			name:             "cpuPerRcu=2: 8rcu -> totalCpu=16 -> xlarge x1",
			catalog:          multiCatalog,
			cpuPerRcu:        2,
			rcu:              8,
			wantInstanceName: "xlarge",
			wantReplicas:     1,
		},
		{
			name:             "cpuPerRcu=2: 16rcu -> totalCpu=32 -> xlarge x2",
			catalog:          multiCatalog,
			cpuPerRcu:        2,
			rcu:              16,
			wantInstanceName: "xlarge",
			wantReplicas:     2,
		},
	}

	for _, tc := range tests {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			setConfigForTest(t, &config.CentralConfig{
				CDCConfig: config.CDCConfig{
					NewArch: config.CDCNewArchConfig{
						InstanceTypes: tc.catalog,
						CpuPerRcu:     tc.cpuPerRcu,
					},
				},
			})
			gotInst, gotReplicas := selectNewArchTiCDCInstance(tc.rcu)
			require.Equal(t, tc.wantInstanceName, gotInst.Name, "instance type mismatch")
			require.Equal(t, tc.wantReplicas, gotReplicas, "replica count mismatch")
		})
	}
}

func TestBuildEssentialTiCDCCRConfiguresPodResourcesAndAnnotations(t *testing.T) {
	setConfigForTest(t, &config.CentralConfig{
		CDCConfig: config.CDCConfig{
			NewArch: config.CDCNewArchConfig{
				Namespace:           "tidb-serverless",
				UpstreamNamespace:   "tidb-pool",
				UpstreamClusterName: "upstream",
				BaseImage:           "ticdc",
				NodeSelector: map[string]string{
					"serverless.tidbcloud.com/node": "adhoc",
					"adhoc-cpu":                     "16",
					"adhoc-memory":                  "64",
				},
			},
		},
	})
	instType := newArchTiCDCInstanceType{
		Name:          "16c",
		CpuPerReplica: 16,
		MemoryGi:      64,
		StorageGi:     200,
	}

	cr := buildEssentialTiCDCCR(
		"ess-ticdc-test",
		"alicloud",
		"ap-southeast-1",
		instType,
		1,
		"cloud-essd",
		"v8.5.4",
	)

	require.True(t, quantityCmpEq(cr.Spec.TiCDC.Requests[corev1.ResourceCPU], "15"))
	require.True(t, quantityCmpEq(cr.Spec.TiCDC.Limits[corev1.ResourceCPU], "15"))
	require.True(t, quantityCmpEq(cr.Spec.TiCDC.Requests[corev1.ResourceMemory], "58387Mi"))
	require.True(t, quantityCmpEq(cr.Spec.TiCDC.Limits[corev1.ResourceMemory], "58387Mi"))
	require.Equal(t, "false", cr.Spec.TiCDC.Annotations[goatScalerSafeToEvictKey])
	require.NotContains(t, cr.Spec.TiCDC.Annotations, clusterAutoscalerSafeToEvictKey)
	require.True(t, isTiCDCCRSpecMatched(cr, instType, 1))

	awsCR := buildEssentialTiCDCCR(
		"ess-ticdc-test",
		"aws",
		"us-east-1",
		instType,
		1,
		"gp3",
		"v8.5.4",
	)
	require.Equal(t, "false", awsCR.Spec.TiCDC.Annotations[clusterAutoscalerSafeToEvictKey])
	require.NotContains(t, awsCR.Spec.TiCDC.Annotations, goatScalerSafeToEvictKey)
}

func TestApplyEssentialTiCDCCRDesiredSpecBackfillsAnnotations(t *testing.T) {
	existingCR := buildEssentialTiCDCCR(
		"ess-ticdc-test",
		"aws",
		"us-east-1",
		newArchTiCDCInstanceType{},
		1,
		"gp3",
		"v8.5.4",
	)
	existingCR.Spec.TiCDC.Annotations = map[string]string{
		"example.com/existing":   "keep",
		goatScalerSafeToEvictKey: "false",
	}
	desired := buildEssentialTiCDCCR(
		"ess-ticdc-test",
		"aws",
		"us-east-1",
		newArchTiCDCInstanceType{},
		1,
		"gp3",
		"v8.5.4",
	)

	applyEssentialTiCDCCRDesiredSpec(existingCR, desired)

	require.Equal(t, "keep", existingCR.Spec.TiCDC.Annotations["example.com/existing"])
	require.Equal(t, "false", existingCR.Spec.TiCDC.Annotations[clusterAutoscalerSafeToEvictKey])
	require.NotContains(t, existingCR.Spec.TiCDC.Annotations, goatScalerSafeToEvictKey)
}

func TestBuildChangefeedURLUsesKeyspaceName(t *testing.T) {
	t.Parallel()

	client := &newArchTiCDCClient{
		baseURL: "https://example.com/proxy/namespace/tidb-serverless/cluster/serverless-cluster",
	}

	got := client.buildChangefeedURL("ess-ticdc-us-east-1-fb2620c6", "", "mVPMqMuJovrSVHt")
	expected := "https://example.com/proxy/namespace/tidb-serverless/cluster/serverless-cluster/ticdc/" +
		"ess-ticdc-us-east-1-fb2620c6/api/v2/changefeeds?keyspace=mVPMqMuJovrSVHt"

	require.Equal(t, expected, got)
}

func TestGetNewArchKeyspaceName(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	mgmtSvc.EXPECT().
		GetCluster(
			gomock.Any(),
			uint64(10194411360446690844),
			tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
		).
		Return(&tidbmgmtservicev1.Cluster{
			ClusterId: 10194411360446690844,
			ClusterInfo: &tidbmgmtservicev1.Cluster_ServerlessCluster{
				ServerlessCluster: &globalserverlessv1.ServerlessServiceCluster{
					Cluster: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
						UserPrefix: "mVPMqMuJovrSVHt",
						KeyspaceId: 554085,
					},
				},
			},
		}, nil)

	svc := &cdcService{mgmtSvc: mgmtSvc}

	got, err := svc.getNewArchKeyspaceName(context.Background(), "10194411360446690844")

	require.NoError(t, err)
	require.Equal(t, "mVPMqMuJovrSVHt", got)
}

func TestNormalizeCreateChangefeedDefaultsUsesWhitelist(t *testing.T) {
	ctrl := gomock.NewController(t)
	featureGate := mock_facade.NewMockFeatureGate(ctrl)
	cluster := &tidbmgmtservicev1.Cluster{
		OrganizationId: 42,
		Region:         "ap-southeast-1",
	}
	changefeed := &vo.Changefeed{}

	setConfigForTest(t, &config.CentralConfig{
		CDCConfig: config.CDCConfig{
			NewArch: config.CDCNewArchConfig{Enabled: true},
		},
	})
	featureGate.EXPECT().FeatureEnabled(constant.FeatureCDCNewArch, uint64(42), "ap-southeast-1").Return(true)

	svc := &cdcService{featureGate: featureGate}
	svc.normalizeCreateChangefeedDefaults(changefeed, cluster)

	require.Equal(t, constant.ChangefeedArchTiCDCNewArch, changefeed.Arch)
	require.Equal(t, uint32(3), changefeed.RCU)
}

func TestValidateCreateChangefeedDefaultsRejectsNonWhitelistedTenant(t *testing.T) {
	ctrl := gomock.NewController(t)
	featureGate := mock_facade.NewMockFeatureGate(ctrl)
	cluster := &tidbmgmtservicev1.Cluster{
		OrganizationId: 42,
		Region:         "ap-southeast-1",
		CloudProvider:  "alicloud",
	}

	setConfigForTest(t, &config.CentralConfig{
		CDCConfig: config.CDCConfig{
			NewArch: config.CDCNewArchConfig{
				Enabled:    true,
				BaseImage:  "ticdc-image",
				Version:    "v8.5.4",
				IssuerName: "issuer",
				InstanceTypes: []config.CDCNewArchInstanceType{
					{Name: "4c", CpuPerReplica: 4, MemoryGi: 8, StorageGi: 50, Iops: 4000, ThroughputMbs: 400},
				},
			},
		},
	})
	featureGate.EXPECT().FeatureEnabled(constant.FeatureCDCNewArch, uint64(42), "ap-southeast-1").Return(false)

	svc := &cdcService{featureGate: featureGate}
	err := svc.validateCreateChangefeedDefaults(&vo.Changefeed{
		Arch: constant.ChangefeedArchTiCDCNewArch,
		RCU:  2,
	}, cluster)

	require.Error(t, err)
	require.Contains(t, err.Error(), "not enabled for your organization or cluster")
}

func TestIsNewArchAllowedForClusterUsesClusterAllowlist(t *testing.T) {
	cluster := &tidbmgmtservicev1.Cluster{
		ClusterId:      123,
		OrganizationId: 42,
		Region:         "ap-southeast-1",
	}

	setConfigForTest(t, &config.CentralConfig{
		CDCConfig: config.CDCConfig{
			NewArch: config.CDCNewArchConfig{
				Enabled:         true,
				EnabledClusters: []string{"123"},
			},
		},
	})

	svc := &cdcService{}
	require.True(t, svc.isNewArchAllowedForCluster(cluster))
}

func TestIsNewArchAllowedForClusterUsesTenantOrClusterAllowlist(t *testing.T) {
	ctrl := gomock.NewController(t)
	featureGate := mock_facade.NewMockFeatureGate(ctrl)
	cluster := &tidbmgmtservicev1.Cluster{
		ClusterId:      123,
		OrganizationId: 42,
		Region:         "ap-southeast-1",
	}

	setConfigForTest(t, &config.CentralConfig{
		CDCConfig: config.CDCConfig{
			NewArch: config.CDCNewArchConfig{
				Enabled:         true,
				EnabledClusters: []string{"456"},
			},
		},
	})
	featureGate.EXPECT().FeatureEnabled(constant.FeatureCDCNewArch, uint64(42), "ap-southeast-1").Return(true)

	svc := &cdcService{featureGate: featureGate}
	require.True(t, svc.isNewArchAllowedForCluster(cluster))
}

func TestIsNewArchAllowedForClusterRejectsWhenNeitherAllowlistMatches(t *testing.T) {
	ctrl := gomock.NewController(t)
	featureGate := mock_facade.NewMockFeatureGate(ctrl)
	cluster := &tidbmgmtservicev1.Cluster{
		ClusterId:      123,
		OrganizationId: 42,
		Region:         "ap-southeast-1",
	}

	setConfigForTest(t, &config.CentralConfig{
		CDCConfig: config.CDCConfig{
			NewArch: config.CDCNewArchConfig{
				Enabled:         true,
				EnabledClusters: []string{"456"},
			},
		},
	})
	featureGate.EXPECT().FeatureEnabled(constant.FeatureCDCNewArch, uint64(42), "ap-southeast-1").Return(false)

	svc := &cdcService{featureGate: featureGate}
	require.False(t, svc.isNewArchAllowedForCluster(cluster))
}

// TestCleanupNewArchTiCDCDataplane verifies the delete-side cleanup
// of the per-shard cert-manager Certificates + the StatefulSet PVCs
// that tidb-operator leaves behind when the TidbCluster is removed.
//
// It pins down the contract:
//   - both Certificates (`-cluster-client-secret`, `-ticdc-cluster-secret`)
//     are deleted by exact name in cdc.newArch.namespace;
//   - PVCs are probed in ordinal order; the first NotFound stops the loop
//     so we don't issue 16 wasted Delete calls on a 1-replica shard.
//
// A real 404 from the dataplane (a previously failed cleanup mid-flight)
// counts as success, so the routine is idempotent and retry-safe.
func TestCleanupNewArchTiCDCDataplane(t *testing.T) {
	const (
		clusterName = "ess-ticdc-us-east-1-fb2620c6"
		namespace   = "tidb-serverless"
	)
	setConfigForTest(t, &config.CentralConfig{
		CDCConfig: config.CDCConfig{
			NewArch: config.CDCNewArchConfig{Namespace: namespace},
		},
	})

	notFound := &tidbcloud.ResponseError{
		StatusCode: http.StatusNotFound,
		Err:        errors.New("not found"),
	}

	type call struct {
		kind string
		name string
		ret  error
	}
	tests := []struct {
		name  string
		calls []call
	}{
		{
			name: "single-replica happy path: 2 certs + PVC-0 ok, PVC-1 NotFound stops",
			calls: []call{
				{kind: "Certificate", name: clusterName + "-cluster-client-secret"},
				{kind: "Certificate", name: clusterName + "-ticdc-cluster-secret"},
				{kind: "PersistentVolumeClaim", name: "ticdc-sort-kv-" + clusterName + "-ticdc-0"},
				{kind: "PersistentVolumeClaim", name: "ticdc-sort-kv-" + clusterName + "-ticdc-1", ret: notFound},
			},
		},
		{
			name: "certs already gone: NotFound is success, still probes PVCs",
			calls: []call{
				{kind: "Certificate", name: clusterName + "-cluster-client-secret", ret: notFound},
				{kind: "Certificate", name: clusterName + "-ticdc-cluster-secret", ret: notFound},
				{kind: "PersistentVolumeClaim", name: "ticdc-sort-kv-" + clusterName + "-ticdc-0", ret: notFound},
			},
		},
	}
	for _, tc := range tests {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			ctrl := gomock.NewController(t)
			proxy := mock_facade.NewMockServerlessClusterProxyFacade(ctrl)
			for _, c := range tc.calls {
				proxy.EXPECT().
					CallK8sResource(gomock.Any(), gomock.AssignableToTypeOf(&clusterproxy.K8sResourceRequest{})).
					DoAndReturn(func(_ context.Context, req *clusterproxy.K8sResourceRequest) (any, error) {
						require.Equal(t, "Delete", req.OperateType)
						require.Equal(t, c.kind, req.Kind)
						require.Equal(t, c.name, req.Name)
						require.Equal(t, namespace, req.Namespace)
						return nil, c.ret
					})
			}
			svc := &cdcService{}
			require.NoError(t, svc.cleanupNewArchTiCDCDataplane(context.Background(), proxy, clusterName))
		})
	}
}

func TestMarkNewArchTiCDCClusterRunningAcksResources(t *testing.T) {
	t.Parallel()

	clusterRepo := &fakeTiCDCClusterRepo{}
	resourceRepo := &fakeTiCDCChangefeedResourceRepo{}
	resources := []*entity.TiCDCChangefeedResource{
		{ClusterID: "c1", ChangefeedID: "cf1", ExpectedSpecification: "2rcu", ActualSpecification: ""},
		{ClusterID: "c2", ChangefeedID: "cf2", ExpectedSpecification: "", ActualSpecification: "2rcu"},
	}
	svc := &cdcService{
		tiCDCClusterRepo:            clusterRepo,
		tiCDCChangefeedResourceRepo: resourceRepo,
	}

	err := svc.markNewArchTiCDCClusterRunning(
		context.Background(),
		&newArchShard{
			Provider:    "aws",
			Region:      "us-east-1",
			Kubernetes:  "k8s-a",
			ShareScope:  "cluster",
			ShareKey:    "c1",
			ClusterName: "ess-ticdc-us-east-1-test",
		},
		"2rcu",
		"v8.5.4",
		resources,
	)

	require.NoError(t, err)
	require.Equal(t, constant.TiCDCClusterStateRunning, clusterRepo.cluster.State)
	require.Equal(t, "2rcu", clusterRepo.cluster.ExpectedSpec)
	require.Equal(t, "2rcu", clusterRepo.cluster.ActualSpec)
	require.Equal(t, "2rcu", resources[0].ActualSpecification)
	require.Empty(t, resources[1].ActualSpecification)
}

func TestMarkNewArchTiCDCClusterDeletedReleasesResources(t *testing.T) {
	t.Parallel()

	clusterRepo := &fakeTiCDCClusterRepo{}
	resourceRepo := &fakeTiCDCChangefeedResourceRepo{}
	resources := []*entity.TiCDCChangefeedResource{
		{ClusterID: "c1", ChangefeedID: "cf1", ExpectedSpecification: "", ActualSpecification: "8rcu"},
		{ClusterID: "c2", ChangefeedID: "cf2", ExpectedSpecification: "", ActualSpecification: ""},
	}
	svc := &cdcService{
		tiCDCClusterRepo:            clusterRepo,
		tiCDCChangefeedResourceRepo: resourceRepo,
	}

	err := svc.markNewArchTiCDCClusterDeleted(
		context.Background(),
		&newArchShard{
			Provider:    "alicloud",
			Region:      "ap-southeast-1",
			Kubernetes:  "dev-ap-southeast-1-f01",
			ShareScope:  "tenant",
			ShareKey:    "t1",
			ClusterName: "ess-ticdc-ap-southeast-1-test",
		},
		"v8.5.4",
		resources,
	)

	require.NoError(t, err)
	require.Equal(t, constant.TiCDCClusterStateDeleted, clusterRepo.cluster.State)
	require.Empty(t, clusterRepo.cluster.ExpectedSpec)
	require.Empty(t, clusterRepo.cluster.ActualSpec)
	require.NotNil(t, clusterRepo.cluster.DeletedAt)
	require.Empty(t, resources[0].ActualSpecification)
	require.Empty(t, resources[1].ActualSpecification)
}

func TestResolveShardVersionUsesRegionalConfigForDeletedShard(t *testing.T) {
	oldConfig := config.Global().CDCConfig.NewArch
	config.Global().CDCConfig.NewArch = config.CDCNewArchConfig{
		Version: "global-version",
		RegionConfig: []config.CDCNewArchRegionalConfig{{
			Provider: "alicloud",
			Region:   "ap-southeast-1",
			Version:  "region-version",
		}},
	}
	defer func() { config.Global().CDCConfig.NewArch = oldConfig }()

	now := time.Now()
	svc := &cdcService{
		tiCDCClusterRepo: &fakeTiCDCClusterRepo{cluster: &entity.TiCDCCluster{
			CloudProvider: "alicloud",
			Region:        "ap-southeast-1",
			Kubernetes:    "k8s-1",
			ShareScope:    "global",
			ShareKey:      "global",
			Version:       "old-version",
			State:         constant.TiCDCClusterStateDeleted,
			DeletedAt:     &now,
		}},
	}

	got := svc.resolveShardVersion(context.Background(), &newArchShard{
		Provider:   "alicloud",
		Region:     "ap-southeast-1",
		Kubernetes: "k8s-1",
		ShareScope: "global",
		ShareKey:   "global",
	})
	require.Equal(t, "region-version", got)
}

func TestSaveNewArchTiCDCClusterRefreshesVersionForDeletedShard(t *testing.T) {
	now := time.Now()
	repo := &fakeTiCDCClusterRepo{cluster: &entity.TiCDCCluster{
		CloudProvider: "alicloud",
		Region:        "ap-southeast-1",
		Kubernetes:    "k8s-1",
		ShareScope:    "global",
		ShareKey:      "global",
		Version:       "old-version",
		State:         constant.TiCDCClusterStateDeleted,
		DeletedAt:     &now,
	}}
	svc := &cdcService{tiCDCClusterRepo: repo}

	err := svc.saveNewArchTiCDCCluster(
		context.Background(),
		&newArchShard{
			Provider:    "alicloud",
			Region:      "ap-southeast-1",
			Kubernetes:  "k8s-1",
			ShareScope:  "global",
			ShareKey:    "global",
			ClusterName: "cluster-name",
		},
		"4rcu",
		"",
		"region-version",
		constant.TiCDCClusterStateCreating,
	)
	require.Nil(t, err)
	require.Equal(t, "region-version", repo.cluster.Version)
	require.Equal(t, constant.TiCDCClusterStateCreating, repo.cluster.State)
	require.Nil(t, repo.cluster.DeletedAt)
}

type fakeTiCDCClusterRepo struct {
	cluster *entity.TiCDCCluster
}

func (r *fakeTiCDCClusterRepo) Save(_ context.Context, cluster *entity.TiCDCCluster) bizerr.BizError {
	copied := *cluster
	r.cluster = &copied
	return nil
}

func (r *fakeTiCDCClusterRepo) GetByShard(
	_ context.Context,
	provider, region, kubernetes, shareScope, shareKey string,
) (*entity.TiCDCCluster, bizerr.BizError) {
	if r.cluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
	}
	if r.cluster.CloudProvider == provider &&
		r.cluster.Region == region &&
		r.cluster.Kubernetes == kubernetes &&
		r.cluster.ShareScope == shareScope &&
		r.cluster.ShareKey == shareKey {
		copied := *r.cluster
		return &copied, nil
	}
	return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
}

func (r *fakeTiCDCClusterRepo) List(
	_ context.Context,
	_, _, _ string,
) ([]*entity.TiCDCCluster, bizerr.BizError) {
	if r.cluster == nil {
		return nil, nil
	}
	return []*entity.TiCDCCluster{r.cluster}, nil
}

type fakeTiCDCChangefeedResourceRepo struct {
	resource *entity.TiCDCChangefeedResource
}

func (r *fakeTiCDCChangefeedResourceRepo) Save(
	_ context.Context,
	resource *entity.TiCDCChangefeedResource,
) bizerr.BizError {
	copied := *resource
	r.resource = &copied
	return nil
}

func (r *fakeTiCDCChangefeedResourceRepo) GetByChangefeedID(
	_ context.Context,
	clusterID, changefeedID string,
) (*entity.TiCDCChangefeedResource, bizerr.BizError) {
	if r.resource != nil &&
		r.resource.ClusterID == clusterID &&
		r.resource.ChangefeedID == changefeedID {
		copied := *r.resource
		return &copied, nil
	}
	return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
}

func (r *fakeTiCDCChangefeedResourceRepo) ListByClusterID(
	_ context.Context,
	clusterID string,
) ([]*entity.TiCDCChangefeedResource, bizerr.BizError) {
	if r.resource != nil && r.resource.ClusterID == clusterID {
		copied := *r.resource
		return []*entity.TiCDCChangefeedResource{&copied}, nil
	}
	return nil, nil
}

func (r *fakeTiCDCChangefeedResourceRepo) ListByShard(
	_ context.Context,
	_, _, _, _, _ string,
) ([]*entity.TiCDCChangefeedResource, bizerr.BizError) {
	return nil, nil
}

func (r *fakeTiCDCChangefeedResourceRepo) ListNeedingSync(
	_ context.Context,
) ([]*entity.TiCDCChangefeedResource, bizerr.BizError) {
	return nil, nil
}

func (r *fakeTiCDCChangefeedResourceRepo) DeleteStaleReleasedRows(
	_ context.Context,
	_ time.Duration,
) (int64, bizerr.BizError) {
	return 0, nil
}

type fakeKernelTiCDCChangefeedRepo struct {
	record *entity.KernelTiCDCChangefeed
}

func (r *fakeKernelTiCDCChangefeedRepo) ListByClusterID(
	_ context.Context,
	clusterID string,
) ([]*entity.KernelTiCDCChangefeed, bizerr.BizError) {
	if r.record != nil && r.record.ClusterID == clusterID {
		copied := *r.record
		return []*entity.KernelTiCDCChangefeed{&copied}, nil
	}
	return nil, nil
}

func (r *fakeKernelTiCDCChangefeedRepo) Save(
	_ context.Context,
	record *entity.KernelTiCDCChangefeed,
) bizerr.BizError {
	copied := *record
	r.record = &copied
	return nil
}

func (r *fakeKernelTiCDCChangefeedRepo) GetByChangefeedID(
	_ context.Context,
	clusterID, changefeedID string,
) (*entity.KernelTiCDCChangefeed, bizerr.BizError) {
	if r.record != nil &&
		r.record.ClusterID == clusterID &&
		r.record.ChangefeedID == changefeedID {
		copied := *r.record
		return &copied, nil
	}
	return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
}

func TestGetChangefeedTiCDCClusterNames(t *testing.T) {
	ctrl := gomock.NewController(t)
	changefeedRepo := mock_repo.NewMockChangefeedRepo(ctrl)
	changefeedRepo.EXPECT().
		ListByFilter(gomock.Any(), entity.ListChangefeedsFilter{ClusterID: "1001"}).
		Return([]*entity.CDCChangefeed{
			{
				ID: "cf-res", ClusterID: "1001", Arch: constant.ChangefeedArchTiCDCNewArch,
				CloudProvider: "aws", Region: "us-east-1", Kubernetes: "k8s-1",
			},
			{
				ID: "cf-kernel", ClusterID: "1001", Arch: constant.ChangefeedArchTiCDCNewArch,
				CloudProvider: "aws", Region: "us-east-1", Kubernetes: "k8s-1",
			},
			{
				ID: "cf-derived", ClusterID: "1001", Arch: constant.ChangefeedArchTiCDCNewArch,
				CloudProvider: "aws", Region: "us-east-1", Kubernetes: "k8s-1",
			},
			{ID: "cf-legacy", ClusterID: "1001", Arch: constant.ChangefeedArchReplicationWorker},
		}, nil)

	svc := &cdcService{
		changefeedRepo: changefeedRepo,
		tiCDCChangefeedResourceRepo: &fakeTiCDCChangefeedResourceRepo{
			resource: &entity.TiCDCChangefeedResource{
				ClusterID: "1001", ChangefeedID: "cf-res", TiCDCClusterName: "from-resource",
				CloudProvider: "aws", Region: "us-east-1", Kubernetes: "k8s-1",
			},
		},
		tiCDCChangefeedRepo: &fakeKernelTiCDCChangefeedRepo{
			record: &entity.KernelTiCDCChangefeed{
				ClusterID: "1001", ChangefeedID: "cf-kernel", TiCDCClusterName: "from-kernel",
				CloudProvider: "aws", Region: "us-east-1", Kubernetes: "k8s-1",
			},
		},
	}

	names, err := svc.GetChangefeedTiCDCClusterNames(
		context.Background(), "1001", []string{"cf-res", "cf-kernel", "cf-derived", "cf-legacy", "cf-unknown"})
	require.Nil(t, err)
	// Recorded assignments win over the config-derived shard.
	require.Equal(t, "from-resource", names["cf-res"])
	require.Equal(t, "from-kernel", names["cf-kernel"])
	// No recorded assignment: fall back to the config-derived name.
	require.Contains(t, names["cf-derived"], "ess-ticdc-")
	// Legacy and unknown changefeeds are absent from the result.
	require.NotContains(t, names, "cf-legacy")
	require.NotContains(t, names, "cf-unknown")
}

func TestGetChangefeedTiCDCClusterNames_PropagatesLookupError(t *testing.T) {
	ctrl := gomock.NewController(t)
	changefeedRepo := mock_repo.NewMockChangefeedRepo(ctrl)
	changefeedRepo.EXPECT().
		ListByFilter(gomock.Any(), gomock.Any()).
		Return(nil, bizerr.New(bizerr.ErrCodeUnknown, "db down"))

	svc := &cdcService{changefeedRepo: changefeedRepo}

	_, err := svc.GetChangefeedTiCDCClusterNames(context.Background(), "1001", []string{"cf-1"})
	require.NotNil(t, err)
}

func TestGetChangefeedTiCDCClusterNames_EmptyInput(t *testing.T) {
	svc := &cdcService{}
	names, err := svc.GetChangefeedTiCDCClusterNames(context.Background(), "1001", nil)
	require.Nil(t, err)
	require.Empty(t, names)
}

func TestIsNewArchTiCDCCRConflictErr(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name string
		err  error
		want bool
	}{
		{name: "nil", err: nil, want: false},
		{
			name: "object modified",
			err: &tidbcloud.ResponseError{
				StatusCode: http.StatusConflict,
				Err:        errors.New("the object has been modified; please apply your changes to the latest version"),
			},
			want: true,
		},
		{
			name: "already exists",
			err: &tidbcloud.ResponseError{
				StatusCode: http.StatusConflict,
				Err:        errors.New("object already exists"),
			},
			want: false,
		},
		{
			name: "wrapped conflict",
			err: bizerr.WrapError(
				errors.New("please apply your changes to the latest version"),
				"update ticdc cr failed",
			),
			want: true,
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			require.Equal(t, tc.want, isNewArchTiCDCCRConflictErr(tc.err))
		})
	}
}

func newUpdateTestCR() *v1alpha1.TidbCluster {
	return buildEssentialTiCDCCR(
		"ess-ticdc-test", "aws", "us-east-1", newArchTiCDCInstanceType{}, 1, "gp3", "v8.5.4",
	)
}

func TestApplyNewArchTiCDCCRDesiredSpecRetriesOnConflict(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	proxy := mock_facade.NewMockServerlessClusterProxyFacade(ctrl)
	cr := newUpdateTestCR()
	desired := newUpdateTestCR()
	desired.Spec.TiCDC.Replicas = 3
	conflict := &tidbcloud.ResponseError{
		StatusCode: http.StatusConflict,
		Err:        errors.New("the object has been modified; please apply your changes to the latest version"),
	}

	updates := 0
	gets := 0
	proxy.EXPECT().CallK8sResource(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, req *clusterproxy.K8sResourceRequest) (any, error) {
			switch req.OperateType {
			case "Get":
				gets++
				latest := newUpdateTestCR()
				latest.ResourceVersion = "2"
				return latest, nil
			case "Update":
				updates++
				updated, err := fromK8sResourceTiCDCCR(req.Object)
				require.NoError(t, err)
				require.Equal(t, int32(3), updated.Spec.TiCDC.Replicas)
				if updates == 1 {
					return nil, conflict
				}
				return nil, nil
			}
			return nil, nil
		}).AnyTimes()

	svc := &cdcService{}
	err := svc.applyNewArchTiCDCCRDesiredSpec(context.Background(), proxy, "ess-ticdc-test", cr, desired)
	require.Nil(t, err)
	require.Equal(t, 2, updates)
	require.Equal(t, 1, gets)
}

func TestApplyNewArchTiCDCCRDesiredSpecStopsOnNonConflict(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	proxy := mock_facade.NewMockServerlessClusterProxyFacade(ctrl)
	fatal := &tidbcloud.ResponseError{StatusCode: http.StatusInternalServerError, Err: errors.New("boom")}
	proxy.EXPECT().CallK8sResource(gomock.Any(), gomock.Any()).Return(nil, fatal).Times(1)

	svc := &cdcService{}
	err := svc.applyNewArchTiCDCCRDesiredSpec(
		context.Background(), proxy, "ess-ticdc-test", newUpdateTestCR(), newUpdateTestCR(),
	)
	require.NotNil(t, err)
}

func TestApplyNewArchTiCDCCRDesiredSpecGivesUpAfterMaxAttempts(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	proxy := mock_facade.NewMockServerlessClusterProxyFacade(ctrl)
	conflict := &tidbcloud.ResponseError{
		StatusCode: http.StatusConflict,
		Err:        errors.New("the object has been modified; please apply your changes to the latest version"),
	}
	updates := 0
	proxy.EXPECT().CallK8sResource(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, req *clusterproxy.K8sResourceRequest) (any, error) {
			if req.OperateType == "Update" {
				updates++
				return nil, conflict
			}
			return newUpdateTestCR(), nil
		}).AnyTimes()

	svc := &cdcService{}
	err := svc.applyNewArchTiCDCCRDesiredSpec(
		context.Background(), proxy, "ess-ticdc-test", newUpdateTestCR(), newUpdateTestCR(),
	)
	require.NotNil(t, err)
	require.Equal(t, newArchTiCDCCRUpdateMaxAttempts, updates)
}