package service

//nolint:gosec,revive // buildEssentialTiCDCClusterName must preserve legacy sha1-based shard naming.
import (
	"context"
	"crypto/sha1"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"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"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
)

const (
	tiCDCResourceAPIVersion         = "pingcap.com/v1alpha1"
	tiCDCResourceKind               = "TidbCluster"
	tiCDCDefaultDirName             = "sort-kv"
	tiCDCDefaultDirPath             = "/var/lib/ticdc"
	defaultServerlessSvc            = "serverless-cluster"
	reservedMemoryInMiB             = 1536
	clusterAutoscalerSafeToEvictKey = "cluster-autoscaler.kubernetes.io/safe-to-evict"
	goatScalerSafeToEvictKey        = "goatscaler.io/safe-to-evict"
)

var newArchMemTaxTable = []struct {
	boundary       uint64
	rate           float64
	quickDeduction uint64
}{
	{
		boundary:       1 * 1024,
		rate:           0.0,
		quickDeduction: 255,
	},
	{
		boundary:       4 * 1024,
		rate:           0.25,
		quickDeduction: 0,
	},
	{
		boundary:       8 * 1024,
		rate:           0.2,
		quickDeduction: 205,
	},
	{
		boundary:       16 * 1024,
		rate:           0.1,
		quickDeduction: 1,
	},
	{
		boundary:       128 * 1024,
		rate:           0.06,
		quickDeduction: 1680,
	},
	{
		boundary:       math.MaxInt64,
		rate:           0.02,
		quickDeduction: 6922,
	},
}

// newArchTiCDCInstanceType is a local alias kept so existing call sites
// don't have to learn about the config package. The catalog of cdc
// machine shapes lives in config.NewArchInstanceTypes(); the picker
// (selectNewArchTiCDCInstance) returns one entry plus a replica count.
type newArchTiCDCInstanceType = config.CDCNewArchInstanceType

const (
	newArchShareScopeGlobal     = "global"
	newArchShareScopeTenant     = "tenant"
	newArchShareScopeCluster    = "cluster"
	newArchShareScopeChangefeed = "changefeed"
)

type newArchShard struct {
	Provider    string
	Region      string
	Kubernetes  string
	ShareScope  string
	ShareKey    string
	ClusterName string
}

func resolveNewArchShard(changefeed *entity.CDCChangefeed) newArchShard {
	scope := resolveNewArchShareScope(changefeed)
	key := newArchShareKey(scope, changefeed)
	shard := newArchShard{
		Provider:   changefeed.CloudProvider,
		Region:     changefeed.Region,
		Kubernetes: changefeed.Kubernetes,
		ShareScope: scope,
		ShareKey:   key,
	}
	shard.ClusterName = buildEssentialTiCDCClusterName(&shard)
	return shard
}

func resolveNewArchShareScope(changefeed *entity.CDCChangefeed) string {
	cfg := config.Global().CDCConfig.NewArch
	scope := cfg.ShareScope
	for _, rule := range cfg.ShareRules {
		if !newArchShareRuleMatches(rule, changefeed) {
			continue
		}
		if newArchShareScopePriority(rule.ShareScope) > newArchShareScopePriority(scope) {
			scope = rule.ShareScope
		}
	}
	if scope == "" {
		scope = newArchShareScopeGlobal
	}
	return scope
}

func newArchShareRuleMatches(rule config.CDCNewArchShareRule, changefeed *entity.CDCChangefeed) bool {
	if len(rule.ClusterIDs) != 0 && stringSliceContains(rule.ClusterIDs, changefeed.ClusterID) {
		return true
	}
	if len(rule.TenantIDs) != 0 && stringSliceContains(rule.TenantIDs, changefeed.TenantID) {
		return true
	}
	return false
}

func newArchShareScopePriority(scope string) int {
	switch scope {
	case newArchShareScopeChangefeed:
		return 3
	case newArchShareScopeCluster:
		return 2
	case newArchShareScopeTenant:
		return 1
	default:
		return 0
	}
}

func stringSliceContains(items []string, want string) bool {
	for _, item := range items {
		if item == want {
			return true
		}
	}
	return false
}

func newArchShareKey(scope string, changefeed *entity.CDCChangefeed) string {
	key := newArchShareScopeGlobal
	switch scope {
	case newArchShareScopeTenant:
		key = changefeed.TenantID
	case newArchShareScopeCluster:
		key = changefeed.ClusterID
	case newArchShareScopeChangefeed:
		key = changefeed.ID
	}
	return key
}

func shardFromResource(resource *entity.TiCDCChangefeedResource) newArchShard {
	shard := newArchShard{
		Provider:    resource.CloudProvider,
		Region:      resource.Region,
		Kubernetes:  resource.Kubernetes,
		ShareScope:  resource.ShareScope,
		ShareKey:    resource.ShareKey,
		ClusterName: resource.TiCDCClusterName,
	}
	if shard.ShareScope == "" {
		shard.ShareScope = newArchShareScopeGlobal
	}
	if shard.ShareKey == "" {
		shard.ShareKey = newArchShareScopeGlobal
	}
	if shard.ClusterName == "" {
		shard.ClusterName = buildEssentialTiCDCClusterName(&shard)
	}
	return shard
}

func shardFromKernelRecord(record *entity.KernelTiCDCChangefeed) newArchShard {
	shard := newArchShard{
		Provider:    record.CloudProvider,
		Region:      record.Region,
		Kubernetes:  record.Kubernetes,
		ShareScope:  record.ShareScope,
		ShareKey:    record.ShareKey,
		ClusterName: record.TiCDCClusterName,
	}
	if shard.ShareScope == "" {
		shard.ShareScope = newArchShareScopeGlobal
	}
	if shard.ShareKey == "" {
		shard.ShareKey = newArchShareScopeGlobal
	}
	if shard.ClusterName == "" {
		shard.ClusterName = buildEssentialTiCDCClusterName(&shard)
	}
	return shard
}

func normalizeNewArchShard(provider, region, kubernetes, shareScope, shareKey string) newArchShard {
	if shareScope == "" {
		shareScope = newArchShareScopeGlobal
	}
	if shareKey == "" {
		shareKey = newArchShareScopeGlobal
	}
	shard := newArchShard{
		Provider:   provider,
		Region:     region,
		Kubernetes: kubernetes,
		ShareScope: shareScope,
		ShareKey:   shareKey,
	}
	shard.ClusterName = buildEssentialTiCDCClusterName(&shard)
	return shard
}

func buildEssentialTiCDCClusterName(shard *newArchShard) string {
	seed := fmt.Sprintf("%s:%s:%s:%s:%s",
		shard.Provider, shard.Region, shard.Kubernetes, shard.ShareScope, shard.ShareKey)
	//nolint:gosec // Preserve legacy shard hash output for existing TiCDC cluster names.
	sum := sha1.Sum([]byte(seed))
	return fmt.Sprintf("ess-ticdc-%s-%s", strings.ReplaceAll(shard.Region, "_", "-"), hex.EncodeToString(sum[:])[:8])
}

// GetChangefeedTiCDCClusterNames resolves the shared TiCDC cluster names for
// the given changefeeds of one user cluster with three bulk queries (the
// changefeed rows plus the recorded assignments in the resource and kernel
// tables), instead of per-changefeed lookups. Recorded assignments win over
// the config-derived shard; legacy replication_worker changefeeds are left
// out of the result. Lookup failures are returned, never masked by the
// config-derived fallback.
func (s *cdcService) GetChangefeedTiCDCClusterNames(
	ctx context.Context,
	clusterID string,
	changefeedIDs []string,
) (map[string]string, bizerr.BizError) {
	names := make(map[string]string, len(changefeedIDs))
	if len(changefeedIDs) == 0 {
		return names, nil
	}

	changefeeds, err := s.changefeedRepo.ListByFilter(ctx, entity.ListChangefeedsFilter{ClusterID: clusterID})
	if err != nil {
		return nil, err
	}

	resourceShards := make(map[string]newArchShard)
	if s.tiCDCChangefeedResourceRepo != nil {
		resources, err := s.tiCDCChangefeedResourceRepo.ListByClusterID(ctx, clusterID)
		if err != nil {
			return nil, err
		}
		for _, resource := range resources {
			resourceShards[resource.ChangefeedID] = shardFromResource(resource)
		}
	}
	kernelShards := make(map[string]newArchShard)
	if s.tiCDCChangefeedRepo != nil {
		records, err := s.tiCDCChangefeedRepo.ListByClusterID(ctx, clusterID)
		if err != nil {
			return nil, err
		}
		for _, record := range records {
			kernelShards[record.ChangefeedID] = shardFromKernelRecord(record)
		}
	}

	wanted := make(map[string]struct{}, len(changefeedIDs))
	for _, id := range changefeedIDs {
		wanted[id] = struct{}{}
	}
	for _, changefeed := range changefeeds {
		if _, ok := wanted[changefeed.ID]; !ok || !isNewArchChangefeed(changefeed) {
			continue
		}
		if shard, ok := resourceShards[changefeed.ID]; ok {
			names[changefeed.ID] = shard.ClusterName
		} else if shard, ok := kernelShards[changefeed.ID]; ok {
			names[changefeed.ID] = shard.ClusterName
		} else {
			names[changefeed.ID] = resolveNewArchShard(changefeed).ClusterName
		}
	}
	return names, nil
}

func (s *cdcService) loadNewArchShardForChangefeed(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
) newArchShard {
	if s.tiCDCChangefeedResourceRepo != nil {
		resource, err := s.tiCDCChangefeedResourceRepo.GetByChangefeedID(ctx, changefeed.ClusterID, changefeed.ID)
		if err == nil && resource != nil {
			return shardFromResource(resource)
		}
	}
	if s.tiCDCChangefeedRepo != nil {
		record, err := s.tiCDCChangefeedRepo.GetByChangefeedID(ctx, changefeed.ClusterID, changefeed.ID)
		if err == nil && record != nil {
			return shardFromKernelRecord(record)
		}
	}
	return resolveNewArchShard(changefeed)
}

func convertServerlessChangefeedConfigToV2(req *serverless.ChangefeedConfig) (*clusterproxy.ChangefeedConfig, error) {
	data, err := json.Marshal(req)
	if err != nil {
		return nil, err
	}
	out := &clusterproxy.ChangefeedConfig{}
	return out, json.Unmarshal(data, out)
}

func convertV2CommonInfoToServerless(
	info *clusterproxy.ChangefeedCommonInfo,
) (*serverless.ChangefeedCommonInfo, error) {
	if info == nil {
		return nil, nil
	}
	data, err := json.Marshal(info)
	if err != nil {
		return nil, err
	}
	out := &serverless.ChangefeedCommonInfo{}
	return out, json.Unmarshal(data, out)
}

// selectNewArchTiCDCInstance picks an instance type + replica count
// from the configured catalog for a shard whose aggregated demand is
// `expectedRCU`. The algorithm mirrors the old
// getSatisfiedNewArchTiCDCInstance contract — each catalog entry is
// treated as a tier whose effective rcu ceiling on one replica is
// `CpuPerReplica / cpuPerRcu`:
//
//  1. Sub-smallest demand snaps up to the smallest type, 1 replica.
//  2. Between adjacent tiers, snap to whichever side of the midpoint
//     the demand falls on. Returned as a single replica.
//  3. Exceeding the largest tier's single-replica ceiling: stay on
//     the largest type and pick replicas =
//     round(totalCpu / CpuPerReplica) (round-half-up, matches old).
//
// The empty / zero rcu input is treated as 3 — same fallback the
// original code used.
func selectNewArchTiCDCInstance(expectedRCU uint32) (newArchTiCDCInstanceType, uint32) {
	types := config.NewArchInstanceTypes()
	cpuPerRcu := config.NewArchCpuPerRcu()
	if expectedRCU == 0 {
		expectedRCU = 3
	}

	tierRcu := func(t newArchTiCDCInstanceType) uint32 {
		if cpuPerRcu == 0 {
			return t.CpuPerReplica
		}
		return t.CpuPerReplica / cpuPerRcu
	}

	if expectedRCU <= tierRcu(types[0]) {
		return types[0], 1
	}
	for i := 1; i < len(types); i++ {
		if expectedRCU > tierRcu(types[i]) {
			continue
		}
		boundary := (tierRcu(types[i]) + tierRcu(types[i-1])) / 2
		if expectedRCU < boundary {
			return types[i-1], 1
		}
		return types[i], 1
	}
	last := types[len(types)-1]
	if last.CpuPerReplica == 0 {
		return last, 1
	}
	totalCpu := expectedRCU * cpuPerRcu
	replicas := uint32(math.Floor(float64(totalCpu)/float64(last.CpuPerReplica) + 0.5))
	if replicas == 0 {
		replicas = 1
	}
	return last, replicas
}

// toCoreTolerations converts the yaml-friendly NewArchToleration
// records into k8s corev1.Toleration values. We do the translation
// here (rather than dragging k8s types into the config package) so
// the config layer stays plain-go and mapstructure-friendly.
func toCoreTolerations(in []config.NewArchToleration) []corev1.Toleration {
	if len(in) == 0 {
		return nil
	}
	out := make([]corev1.Toleration, 0, len(in))
	for _, t := range in {
		out = append(out, corev1.Toleration{
			Key:      t.Key,
			Operator: corev1.TolerationOperator(t.Operator),
			Value:    t.Value,
			Effect:   corev1.TaintEffect(t.Effect),
		})
	}
	return out
}

// buildEssentialTiCDCCR builds the TidbCluster CR that ships a shared
// TiCDC cluster for the newarch path.
//
// `version` is the TiCDC image tag to use. Callers MUST pass the
// version they want pinned on the dataplane — typically the value
// already stored in ticdc_clusters.version, falling back to
// config.ResolveNewArchVersion only for the first create of a
// new shard. This indirection is deliberate: a config flip must not
// auto-upgrade every existing shared cluster; upgrades go through
// the explicit ChangefeedOpsService.UpgradeTiCDCCluster ops API.
func buildEssentialTiCDCCR(
	clusterName string,
	provider, region string,
	instType newArchTiCDCInstanceType,
	replicas uint32,
	scName string,
	version string,
) *v1alpha1.TidbCluster {
	cfg := config.Global().CDCConfig.NewArch
	baseImage := config.ResolveNewArchBaseImage(provider, region)
	pvDeletePolicy := corev1.PersistentVolumeReclaimDelete
	cdcConfig := v1alpha1.NewCDCConfig()
	cdcConfig.Set("cluster-id", clusterName)
	cdcConfig.Set("data-dir", tiCDCDefaultDirPath)
	cdcConfig.Set("log-file", "")
	cdcConfig.Set("newarch", true)
	cdcConfig.Set("enable-legacy-safepoint", true)
	var storageClassName *string
	if scName != "" {
		storageClassName = ptr.To(scName)
	}

	storageGi := instType.StorageGi
	resourceList := corev1.ResourceList{}
	if instType.CpuPerReplica > 0 {
		cpuRequest := computeNewArchCPURequest(instType.CpuPerReplica)
		resourceList[corev1.ResourceCPU] = resource.MustParse(fmt.Sprintf("%d", cpuRequest))
	}
	if instType.MemoryGi > 0 {
		memoryRequest := computeNewArchAvailableMemory(instType.MemoryGi)
		resourceList[corev1.ResourceMemory] = resource.MustParse(fmt.Sprintf("%dMi", memoryRequest))
	}
	labels := map[string]string{
		"app.kubernetes.io/managed-by": "serverless-service",
		"tidbcloud.com/cdc-arch":       string(constant.ChangefeedArchTiCDCNewArch),
	}
	if instType.Name != "" {
		labels["tidbcloud.com/cdc-instance-type"] = instType.Name
	}

	tc := &v1alpha1.TidbCluster{
		TypeMeta: metav1.TypeMeta{
			APIVersion: tiCDCResourceAPIVersion,
			Kind:       tiCDCResourceKind,
		},
		ObjectMeta: metav1.ObjectMeta{
			Name:      clusterName,
			Namespace: cfg.Namespace,
			Labels:    labels,
		},
		Spec: v1alpha1.TidbClusterSpec{
			PVReclaimPolicy:      &pvDeletePolicy,
			ConfigUpdateStrategy: v1alpha1.ConfigUpdateStrategyRollingUpdate,
			Version:              version,
			TLSCluster: &v1alpha1.TLSCluster{
				Enabled: true,
			},
			Cluster: &v1alpha1.TidbClusterRef{
				Namespace: cfg.UpstreamNamespace,
				Name:      cfg.UpstreamClusterName,
			},
			TiCDC: &v1alpha1.TiCDCSpec{
				ComponentSpec: v1alpha1.ComponentSpec{
					HostNetwork:        ptr.To(true),
					PodSecurityContext: &corev1.PodSecurityContext{},
					NodeSelector:       cfg.NodeSelector,
					Tolerations:        toCoreTolerations(cfg.Tolerations),
					Annotations: map[string]string{
						safeToEvictAnnotationKey(provider): "false",
					},
				},
				BaseImage: baseImage,
				Replicas:  int32(replicas),
				Config:    cdcConfig,
				ResourceRequirements: corev1.ResourceRequirements{
					Requests: resourceList,
					Limits:   resourceList,
				},
				StorageVolumes: []v1alpha1.StorageVolume{
					{
						Name:             tiCDCDefaultDirName,
						StorageClassName: storageClassName,
						StorageSize:      fmt.Sprintf("%dGi", storageGi),
						MountPath:        tiCDCDefaultDirPath,
					},
				},
			},
		},
	}
	return tc
}

func safeToEvictAnnotationKey(provider string) string {
	if strings.EqualFold(provider, "alicloud") {
		return goatScalerSafeToEvictKey
	}
	return clusterAutoscalerSafeToEvictKey
}

// applyEssentialTiCDCCRDesiredSpec updates the fields managed by this service
// on an existing TidbCluster. Keep create and update paths using the same
// desired CR so newly introduced component settings are also backfilled.
func applyEssentialTiCDCCRDesiredSpec(existingCR, desired *v1alpha1.TidbCluster) {
	existingCR.Spec.Cluster = desired.Spec.Cluster
	existingCR.Spec.TiCDC.BaseImage = desired.Spec.TiCDC.BaseImage
	existingCR.Spec.Version = desired.Spec.Version
	existingCR.Spec.TiCDC.Version = desired.Spec.TiCDC.Version
	existingCR.Spec.TiCDC.Replicas = desired.Spec.TiCDC.Replicas
	existingCR.Spec.TiCDC.Config = desired.Spec.TiCDC.Config
	existingCR.Spec.TiCDC.StorageVolumes = desired.Spec.TiCDC.StorageVolumes
	existingCR.Spec.TiCDC.HostNetwork = desired.Spec.TiCDC.HostNetwork
	existingCR.Spec.TiCDC.NodeSelector = desired.Spec.TiCDC.NodeSelector
	existingCR.Spec.TiCDC.Tolerations = desired.Spec.TiCDC.Tolerations
	if existingCR.Spec.TiCDC.Annotations == nil {
		existingCR.Spec.TiCDC.Annotations = make(map[string]string)
	}
	delete(existingCR.Spec.TiCDC.Annotations, clusterAutoscalerSafeToEvictKey)
	delete(existingCR.Spec.TiCDC.Annotations, goatScalerSafeToEvictKey)
	for key, value := range desired.Spec.TiCDC.Annotations {
		existingCR.Spec.TiCDC.Annotations[key] = value
	}
	existingCR.Spec.TiCDC.ResourceRequirements = desired.Spec.TiCDC.ResourceRequirements
	existingCR.Spec.TLSCluster = desired.Spec.TLSCluster
	existingCR.Spec.ConfigUpdateStrategy = desired.Spec.ConfigUpdateStrategy
	existingCR.Spec.PVReclaimPolicy = desired.Spec.PVReclaimPolicy
}

func formatNewArchSpecification(rcu uint32) string {
	if rcu == 0 {
		rcu = 3
	}
	return fmt.Sprintf("%drcu", rcu)
}

func parseNewArchSpecification(spec string) (uint32, error) {
	spec = strings.TrimSpace(strings.ToLower(spec))
	if spec == "" {
		return 0, nil
	}
	if !strings.HasSuffix(spec, "rcu") {
		return 0, fmt.Errorf("invalid specification: %s", spec)
	}
	value := strings.TrimSuffix(spec, "rcu")
	parsed, err := strconv.ParseUint(value, 10, 32)
	if err != nil {
		return 0, err
	}
	return uint32(parsed), nil
}

func sumNewArchResourceRCU(resources []*entity.TiCDCChangefeedResource) (uint32, error) {
	var total uint32
	for _, resource := range resources {
		if resource == nil || resource.ExpectedSpecification == "" {
			continue
		}
		rcu, err := parseNewArchSpecification(resource.ExpectedSpecification)
		if err != nil {
			return 0, err
		}
		total += rcu
	}
	return total, nil
}

func filterExpectedNewArchResources(
	resources []*entity.TiCDCChangefeedResource,
) []*entity.TiCDCChangefeedResource {
	filtered := make([]*entity.TiCDCChangefeedResource, 0, len(resources))
	for _, resource := range resources {
		if resource == nil || resource.ExpectedSpecification == "" {
			continue
		}
		filtered = append(filtered, resource)
	}
	return filtered
}

func isNewArchChangefeed(changefeed *entity.CDCChangefeed) bool {
	return changefeed != nil && changefeed.Arch == constant.ChangefeedArchTiCDCNewArch
}

func (s *cdcService) getNewArchKeyspaceName(ctx context.Context, clusterID string) (string, bizerr.BizError) {
	if s.mgmtSvc == nil {
		return "", bizerr.New(bizerr.ErrCodeUnknown, "mgmt service is not configured for newarch keyspace lookup")
	}

	uintClusterID, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		return "", bizerr.WrapError(err, "parse cluster id %s", clusterID)
	}

	cluster, bizErr := s.mgmtSvc.GetCluster(ctx, uintClusterID, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if bizErr != nil {
		return "", bizErr
	}
	keyspaceName := cluster.GetServerlessCluster().GetCluster().GetUserPrefix()
	if keyspaceName == "" {
		return "", bizerr.New(bizerr.ErrCodeUnknown, "cluster %s keyspace name is empty", clusterID)
	}
	return keyspaceName, nil
}

// saveNewArchTiCDCCluster upserts the per-shard ticdc_clusters row.
//
// `version` is only honoured on first insert: once the row exists,
// its `version` is preserved. To change `version` on an existing row
// (i.e. trigger an upgrade) use updateNewArchTiCDCClusterVersion,
// which is the only entry point from the ops upgrade API.
func (s *cdcService) saveNewArchTiCDCCluster(
	ctx context.Context,
	shard *newArchShard,
	expectedSpec, actualSpec, version string,
	state constant.TiCDCClusterState,
) bizerr.BizError {
	if s.tiCDCClusterRepo == nil {
		return nil
	}

	cluster, err := s.tiCDCClusterRepo.GetByShard(
		ctx,
		shard.Provider,
		shard.Region,
		shard.Kubernetes,
		shard.ShareScope,
		shard.ShareKey,
	)
	if err != nil {
		if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return err
		}
		cluster = &entity.TiCDCCluster{
			CloudProvider: shard.Provider,
			Region:        shard.Region,
			Kubernetes:    shard.Kubernetes,
			ShareScope:    shard.ShareScope,
			ShareKey:      shard.ShareKey,
			Version:       version,
		}
	} else if cluster.State == constant.TiCDCClusterStateDeleted || cluster.DeletedAt != nil {
		// A deleted shard row is reused because the shard key is unique.
		// Treat the next create as a fresh shard for config defaults so
		// regional version canaries take effect after full teardown.
		cluster.Version = version
	}

	cluster.TiCDCClusterName = shard.ClusterName
	cluster.ExpectedSpec = expectedSpec
	cluster.ActualSpec = actualSpec
	// Preserve `upgrading` across reconcile passes: the ops upgrade
	// API set it explicitly, and the dataplane CR rolls take time. We
	// only clear `upgrading` when the shard converges to running
	// (expected_spec == actual_spec, handled by the caller passing
	// state=running) or when it's torn down (state=deleted). Any
	// reconcile that tries to flip it back to `creating` is treated
	// as a no-op on the state column so the operator can still see
	// the upgrade in flight.
	if cluster.State != constant.TiCDCClusterStateUpgrading ||
		state != constant.TiCDCClusterStateCreating {
		cluster.State = state
	}
	if state == constant.TiCDCClusterStateDeleted {
		now := time.Now().UTC()
		cluster.DeletedAt = &now
	} else {
		cluster.DeletedAt = nil
	}
	return s.tiCDCClusterRepo.Save(ctx, cluster)
}

func (s *cdcService) markNewArchTiCDCClusterRunning(
	ctx context.Context,
	shard *newArchShard,
	specName, version string,
	resources []*entity.TiCDCChangefeedResource,
) bizerr.BizError {
	if err := s.saveNewArchTiCDCCluster(
		ctx,
		shard,
		specName,
		specName,
		version,
		constant.TiCDCClusterStateRunning,
	); err != nil {
		return err
	}
	return s.markNewArchChangefeedResourcesActualExpected(ctx, resources)
}

func (s *cdcService) markNewArchTiCDCClusterDeleted(
	ctx context.Context,
	shard *newArchShard,
	version string,
	resources []*entity.TiCDCChangefeedResource,
) bizerr.BizError {
	if err := s.saveNewArchTiCDCCluster(
		ctx,
		shard,
		"",
		"",
		version,
		constant.TiCDCClusterStateDeleted,
	); err != nil {
		return err
	}
	return s.markNewArchChangefeedResourcesActualExpected(ctx, resources)
}

func (s *cdcService) markNewArchChangefeedResourcesActualExpected(
	ctx context.Context,
	resources []*entity.TiCDCChangefeedResource,
) bizerr.BizError {
	for _, resource := range resources {
		if resource == nil || resource.ActualSpecification == resource.ExpectedSpecification {
			continue
		}
		resource.ActualSpecification = resource.ExpectedSpecification
		if saveErr := s.tiCDCChangefeedResourceRepo.Save(ctx, resource); saveErr != nil {
			return saveErr
		}
	}
	return nil
}

// resolveShardVersion returns the version to render into the CR for
// the given shard. It honours an existing ticdc_clusters.version
// (set on first create or by an explicit upgrade) and falls back to
// the configured default only when the shard has no row yet.
func (s *cdcService) resolveShardVersion(
	ctx context.Context,
	shard *newArchShard,
) string {
	cfgVersion := config.ResolveNewArchVersion(shard.Provider, shard.Region)
	if s.tiCDCClusterRepo == nil {
		return cfgVersion
	}
	cluster, err := s.tiCDCClusterRepo.GetByShard(
		ctx,
		shard.Provider,
		shard.Region,
		shard.Kubernetes,
		shard.ShareScope,
		shard.ShareKey,
	)
	if err != nil || cluster == nil || cluster.Version == "" ||
		cluster.State == constant.TiCDCClusterStateDeleted || cluster.DeletedAt != nil {
		return cfgVersion
	}
	return cluster.Version
}

// ListNewArchTiCDCClusters — ops surface, see CDCService interface.
func (s *cdcService) ListNewArchTiCDCClusters(
	ctx context.Context,
	provider, region, state string,
) ([]*entity.TiCDCCluster, bizerr.BizError) {
	if s.tiCDCClusterRepo == nil {
		return nil, nil
	}
	return s.tiCDCClusterRepo.List(ctx, provider, region, state)
}

// GetNewArchTiCDCCluster — ops surface, see CDCService interface.
func (s *cdcService) GetNewArchTiCDCCluster(
	ctx context.Context,
	provider, region, kubernetes, shareScope, shareKey string,
) (*entity.TiCDCCluster, bizerr.BizError) {
	if s.tiCDCClusterRepo == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "ticdc cluster repo not configured")
	}
	shard := normalizeNewArchShard(provider, region, kubernetes, shareScope, shareKey)
	return s.tiCDCClusterRepo.GetByShard(
		ctx,
		shard.Provider,
		shard.Region,
		shard.Kubernetes,
		shard.ShareScope,
		shard.ShareKey,
	)
}

// UpgradeNewArchTiCDCCluster sets ticdc_clusters.version = target and
// flips state to upgrading, then triggers reconcile so the next CR
// update reaches the dataplane with the new version. The tidb-operator
// is responsible for the rolling pod replace afterwards.
//
// This is the only call site that mutates ticdc_clusters.version on
// an existing row — saveNewArchTiCDCCluster preserves version on
// upsert by design.
func (s *cdcService) UpgradeNewArchTiCDCCluster(
	ctx context.Context,
	provider, region, kubernetes, shareScope, shareKey, targetVersion string,
) (*entity.TiCDCCluster, bizerr.BizError) {
	if s.tiCDCClusterRepo == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "ticdc cluster repo not configured")
	}
	if targetVersion == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "target_version is required")
	}
	shard := normalizeNewArchShard(provider, region, kubernetes, shareScope, shareKey)
	cluster, err := s.tiCDCClusterRepo.GetByShard(
		ctx,
		shard.Provider,
		shard.Region,
		shard.Kubernetes,
		shard.ShareScope,
		shard.ShareKey,
	)
	if err != nil {
		return nil, err
	}
	if cluster.State == constant.TiCDCClusterStateDeleted {
		return nil, bizerr.New(bizerr.ErrCodeParams,
			"cannot upgrade a deleted ticdc cluster (%s/%s/%s)", provider, region, kubernetes)
	}
	if cluster.Version == targetVersion {
		// No-op upgrade. Don't touch state — the cluster might already
		// be running on this version.
		return cluster, nil
	}
	if err := s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		cluster.Version = targetVersion
		cluster.State = constant.TiCDCClusterStateUpgrading
		return s.tiCDCClusterRepo.Save(ctx, cluster)
	}); err != nil {
		if bErr, ok := err.(bizerr.BizError); ok {
			return nil, bErr
		}
		return nil, bizerr.WrapError(err, "save ticdc cluster upgrade intent")
	}
	// Trigger an immediate reconcile so the CR gets re-rendered with
	// the new version. If this fails (proxy down, transient k8s), the
	// background sync worker will retry — but pushing it inline gives
	// the ops caller a tight feedback loop.
	if recErr := s.reconcileNewArchTiCDCClusterForShard(ctx, &shard); recErr != nil {
		// Log via returning a wrapped error; the DB intent is already
		// written, so the upgrade is in flight either way.
		return cluster, bizerr.WrapError(recErr, "reconcile after upgrade write")
	}
	return cluster, nil
}

func toK8sResourceObject(obj any) (map[string]any, error) {
	data, err := json.Marshal(obj)
	if err != nil {
		return nil, err
	}
	out := make(map[string]any)
	err = json.Unmarshal(data, &out)
	return out, err
}

func fromK8sResourceTiCDCCR(data any) (*v1alpha1.TidbCluster, error) {
	payload, err := json.Marshal(data)
	if err != nil {
		return nil, err
	}
	out := &v1alpha1.TidbCluster{}
	return out, json.Unmarshal(payload, out)
}

func isTiCDCCRReady(cr *v1alpha1.TidbCluster) bool {
	if cr == nil || cr.Spec.TiCDC == nil || cr.Status.TiCDC.StatefulSet == nil {
		return false
	}
	return cr.Status.TiCDC.StatefulSet.ReadyReplicas >= cr.Spec.TiCDC.Replicas
}

// isTiCDCCRSpecMatched returns true when the existing CR's spec
// matches the desired tier. Comparison uses resource.Quantity.Cmp
// rather than String() equality so equivalent renderings — e.g.
// `"8000m"` vs `"8"` for CPU, `"8192Mi"` vs `"8Gi"` for memory —
// don't accidentally read as a spec drift and force an unnecessary
// CR update.
func isTiCDCCRSpecMatched(
	cr *v1alpha1.TidbCluster,
	instType newArchTiCDCInstanceType,
	replicas uint32,
) bool {
	if cr == nil || cr.Spec.TiCDC == nil {
		return false
	}
	if cr.Spec.TiCDC.Replicas != int32(replicas) {
		return false
	}
	if len(cr.Spec.TiCDC.StorageVolumes) == 0 {
		return false
	}
	expectedStorageGi := instType.StorageGi
	if !quantityEqual(cr.Spec.TiCDC.StorageVolumes[0].StorageSize, fmt.Sprintf("%dGi", expectedStorageGi)) {
		return false
	}
	if instType.CpuPerReplica > 0 {
		cpu, ok := cr.Spec.TiCDC.Requests[corev1.ResourceCPU]
		if !ok || !quantityCmpEq(cpu, fmt.Sprintf("%d", computeNewArchCPURequest(instType.CpuPerReplica))) {
			return false
		}
	}
	if instType.MemoryGi > 0 {
		mem, ok := cr.Spec.TiCDC.Requests[corev1.ResourceMemory]
		if !ok || !quantityCmpEq(mem, fmt.Sprintf("%dMi", computeNewArchAvailableMemory(instType.MemoryGi))) {
			return false
		}
	}
	return true
}

func computeNewArchCPURequest(cpuCore uint32) uint32 {
	if cpuCore == 0 {
		return 0
	}
	return cpuCore - 1
}

func computeNewArchAvailableMemory(memoryInGiB uint32) uint64 {
	memoryInMiB := uint64(memoryInGiB) * 1024
	for _, level := range newArchMemTaxTable {
		if memoryInMiB <= level.boundary {
			deduction := uint64(math.Ceil(float64(memoryInMiB)*level.rate)) + level.quickDeduction
			if memoryInMiB <= deduction {
				memoryInMiB = 0
			} else {
				memoryInMiB -= deduction
			}
			break
		}
	}
	if memoryInMiB <= reservedMemoryInMiB {
		memoryInMiB = 0
	} else {
		memoryInMiB -= reservedMemoryInMiB
	}
	return memoryInMiB
}

// quantityEqual parses both strings as resource.Quantity and compares
// them. A parse error on either side falls back to literal string
// equality so we don't crash on hand-edited CRs.
func quantityEqual(a, b string) bool {
	qa, errA := resource.ParseQuantity(a)
	qb, errB := resource.ParseQuantity(b)
	if errA != nil || errB != nil {
		return a == b
	}
	return qa.Cmp(qb) == 0
}

// quantityCmpEq compares an already-parsed Quantity with a string.
func quantityCmpEq(have resource.Quantity, want string) bool {
	w, err := resource.ParseQuantity(want)
	if err != nil {
		return have.String() == want
	}
	return have.Cmp(w) == 0
}

func (s *cdcService) normalizeCreateChangefeedDefaults(
	changefeed *vo.Changefeed,
	cluster *tidbmgmtservicev1.Cluster,
) {
	defaultArch := constant.ChangefeedArchReplicationWorker
	if s.isNewArchAllowedForCluster(cluster) {
		defaultArch = constant.ChangefeedArchTiCDCNewArch
	}
	if changefeed.Arch == "" {
		changefeed.Arch = defaultArch
	}
	if changefeed.RCU == 0 {
		changefeed.RCU = 3
	}
}

func (s *cdcService) validateCreateChangefeedDefaults(
	changefeed *vo.Changefeed,
	cluster *tidbmgmtservicev1.Cluster,
) bizerr.BizError {
	switch changefeed.Arch {
	case constant.ChangefeedArchReplicationWorker:
		return nil
	case constant.ChangefeedArchTiCDCNewArch:
		if err := s.validateNewArchAllowedForCluster(cluster); err != nil {
			return err
		}
		cfg := config.Global().CDCConfig.NewArch
		provider := cluster.CloudProvider
		region := cluster.Region
		if config.ResolveNewArchBaseImage(provider, region) == "" {
			return bizerr.New(bizerr.ErrCodeUnknown,
				"ticdc newarch configuration is incomplete: configure either "+
					"cdc.newArch.regionConfig[].baseImage, cdc.newArch.baseImagePerProvider, "+
					"or cdc.newArch.baseImage for %s/%s",
				provider,
				region)
		}
		if config.ResolveNewArchVersion(provider, region) == "" {
			return bizerr.New(bizerr.ErrCodeUnknown,
				"ticdc newarch configuration is incomplete: configure either "+
					"cdc.newArch.regionConfig[].version or cdc.newArch.version for %s/%s",
				provider,
				region)
		}
		if cfg.IssuerName == "" {
			return bizerr.New(bizerr.ErrCodeUnknown,
				"ticdc newarch configuration is incomplete: cdc.newArch.issuerName is empty")
		}
		if len(config.NewArchInstanceTypes()) == 0 {
			return bizerr.New(bizerr.ErrCodeUnknown,
				"ticdc newarch configuration is incomplete: cdc.newArch.specifications is empty")
		}
		if changefeed.RCU == 0 {
			return bizerr.New(bizerr.ErrCodeParams, "rcu must be greater than 0")
		}
		return nil
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported changefeed arch: %s", changefeed.Arch)
	}
}

func (s *cdcService) isNewArchAllowedForCluster(cluster *tidbmgmtservicev1.Cluster) bool {
	cfg := config.Global().CDCConfig.NewArch
	if !cfg.Enabled || cluster == nil {
		return false
	}
	if cluster.ClusterId != 0 && stringSliceContains(cfg.EnabledClusters, strconv.FormatUint(cluster.ClusterId, 10)) {
		return true
	}
	if s.featureGate == nil {
		return false
	}
	return s.featureGate.FeatureEnabled(constant.FeatureCDCNewArch, cluster.OrganizationId, cluster.Region)
}

func (s *cdcService) validateNewArchAllowedForCluster(cluster *tidbmgmtservicev1.Cluster) bizerr.BizError {
	cfg := config.Global().CDCConfig.NewArch
	if !cfg.Enabled {
		return bizerr.New(bizerr.ErrCodeParams, "ticdc newarch changefeed is disabled")
	}
	if !s.isNewArchAllowedForCluster(cluster) {
		return bizerr.New(bizerr.ErrCodeParams,
			"ticdc newarch changefeed is not enabled for your organization or cluster")
	}
	return nil
}

func (s *cdcService) getNewArchTiCDCClient(
	ctx context.Context,
	provider, region, k8s string,
) (*newArchTiCDCClient, bizerr.BizError) {
	clusterProxyURL := config.GetClusterProxyUrl(provider, region, k8s)
	if clusterProxyURL == "" {
		regionInfo := fmt.Sprintf("%s.%s.%s", provider, region, k8s)
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster proxy url not found in config for: %s", regionInfo)
	}
	cli, err := newNewArchTiCDCClient(ctx, clusterProxyURL)
	if err != nil {
		return nil, bizerr.WrapError(err, "build newarch ticdc client failed")
	}
	return cli, nil
}

func (s *cdcService) getNewArchChangefeedResource(
	ctx context.Context,
	clusterID, changefeedID string,
) (*entity.TiCDCChangefeedResource, bizerr.BizError) {
	return s.tiCDCChangefeedResourceRepo.GetByChangefeedID(ctx, clusterID, changefeedID)
}

func (s *cdcService) ensureNewArchChangefeedResource(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
	expectedSpecification string,
) (*newArchShard, bizerr.BizError) {
	resource, err := s.getNewArchChangefeedResource(ctx, changefeed.ClusterID, changefeed.ID)
	shard := resolveNewArchShard(changefeed)
	if err != nil {
		if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return nil, err
		}
		resource = &entity.TiCDCChangefeedResource{
			ClusterID:        changefeed.ClusterID,
			ChangefeedID:     changefeed.ID,
			CloudProvider:    shard.Provider,
			Region:           shard.Region,
			Kubernetes:       shard.Kubernetes,
			ShareScope:       shard.ShareScope,
			ShareKey:         shard.ShareKey,
			TiCDCClusterName: shard.ClusterName,
		}
	} else {
		shard = shardFromResource(resource)
	}
	resource.CloudProvider = shard.Provider
	resource.Region = shard.Region
	resource.Kubernetes = shard.Kubernetes
	resource.ShareScope = shard.ShareScope
	resource.ShareKey = shard.ShareKey
	resource.TiCDCClusterName = shard.ClusterName
	resource.ExpectedSpecification = expectedSpecification
	if err := s.tiCDCChangefeedResourceRepo.Save(ctx, resource); err != nil {
		return nil, err
	}
	return &shard, nil
}

func (s *cdcService) ensureNewArchTiCDCChangefeedRecord(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
	shard *newArchShard,
	kernelStatus constant.TiCDCChangefeedKernelStatus,
) bizerr.BizError {
	record, err := s.tiCDCChangefeedRepo.GetByChangefeedID(ctx, changefeed.ClusterID, changefeed.ID)
	if err != nil {
		if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return err
		}
		record = &entity.KernelTiCDCChangefeed{
			ClusterID:          changefeed.ClusterID,
			ChangefeedID:       changefeed.ID,
			CloudProvider:      shard.Provider,
			Region:             shard.Region,
			Kubernetes:         shard.Kubernetes,
			ShareScope:         shard.ShareScope,
			ShareKey:           shard.ShareKey,
			KeySpaceID:         changefeed.KeySpaceID,
			TiCDCClusterName:   shard.ClusterName,
			KernelChangefeedID: changefeed.ID,
		}
	}
	record.CloudProvider = shard.Provider
	record.Region = shard.Region
	record.Kubernetes = shard.Kubernetes
	record.ShareScope = shard.ShareScope
	record.ShareKey = shard.ShareKey
	record.KeySpaceID = changefeed.KeySpaceID
	record.TiCDCClusterName = shard.ClusterName
	record.KernelChangefeedID = changefeed.ID
	record.KernelStatus = kernelStatus
	if kernelStatus == constant.TiCDCChangefeedKernelStatusDeleted {
		now := time.Now().UTC()
		record.DeletedAt = &now
	} else {
		record.DeletedAt = nil
	}
	return s.tiCDCChangefeedRepo.Save(ctx, record)
}

func (s *cdcService) persistNewArchTiCDCChangefeedCreateIntent(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
	shard *newArchShard,
) (*entity.KernelTiCDCChangefeed, *newArchShard, bizerr.BizError) {
	record, err := s.tiCDCChangefeedRepo.GetByChangefeedID(ctx, changefeed.ClusterID, changefeed.ID)
	if err != nil {
		if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return nil, nil, err
		}
		record = &entity.KernelTiCDCChangefeed{
			ClusterID:          changefeed.ClusterID,
			ChangefeedID:       changefeed.ID,
			KernelChangefeedID: changefeed.ID,
			KernelStatus:       constant.TiCDCChangefeedKernelStatusPending,
		}
	}
	if record.KernelStatus == constant.TiCDCChangefeedKernelStatusActive {
		recordShard := shardFromKernelRecord(record)
		return record, &recordShard, nil
	}
	if record.TiCDCClusterName != "" {
		recordShard := shardFromKernelRecord(record)
		shard = &recordShard
	}
	record.CloudProvider = shard.Provider
	record.Region = shard.Region
	record.Kubernetes = shard.Kubernetes
	record.ShareScope = shard.ShareScope
	record.ShareKey = shard.ShareKey
	record.KeySpaceID = changefeed.KeySpaceID
	record.TiCDCClusterName = shard.ClusterName
	record.KernelChangefeedID = changefeed.ID
	record.KernelStatus = constant.TiCDCChangefeedKernelStatusPending
	record.DeletedAt = nil
	if err := s.tiCDCChangefeedRepo.Save(ctx, record); err != nil {
		return nil, nil, err
	}
	return record, shard, nil
}

func (s *cdcService) listNewArchChangefeedResourcesForShard(
	ctx context.Context,
	shard *newArchShard,
) ([]*entity.TiCDCChangefeedResource, bizerr.BizError) {
	return s.tiCDCChangefeedResourceRepo.ListByShard(
		ctx,
		shard.Provider,
		shard.Region,
		shard.Kubernetes,
		shard.ShareScope,
		shard.ShareKey,
	)
}

func (s *cdcService) isNewArchChangefeedResourceReady(
	ctx context.Context,
	clusterID, changefeedID string,
) (bool, bizerr.BizError) {
	resource, err := s.getNewArchChangefeedResource(ctx, clusterID, changefeedID)
	if err != nil {
		return false, err
	}
	return resource.ExpectedSpecification != "" && resource.ExpectedSpecification == resource.ActualSpecification, nil
}

func (s *cdcService) isNewArchChangefeedResourceReleased(
	ctx context.Context,
	clusterID, changefeedID string,
) (bool, bizerr.BizError) {
	resource, err := s.getNewArchChangefeedResource(ctx, clusterID, changefeedID)
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return true, nil
		}
		return false, err
	}
	return resource.ExpectedSpecification == "" && resource.ActualSpecification == "", nil
}

// isAlreadyExistsErr reports whether a clusterproxy CallK8sResource
// "Create" error means the target object already exists. This is how
// we make Create calls idempotent in the face of partial-failure
// retries: a previous reconcile may have created the object but
// failed to record DB state, or two reconciles may race.
//
// HTTP 409 Conflict is the standard k8s api-server response for
// "already exists"; the clusterproxy passes it through verbatim.
func isAlreadyExistsErr(err error) bool {
	if err == nil {
		return false
	}
	var respErr *tidbcloud.ResponseError
	if errors.As(err, &respErr) && respErr.StatusCode == http.StatusConflict {
		return true
	}
	// Fallback string match in case the proxy ever wraps the status
	// in something less structured.
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "already exists")
}

func isNewArchChangefeedAlreadyExistsErr(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "errchangefeedalreadyexists") ||
		strings.Contains(msg, "changefeed already exists") ||
		strings.Contains(msg, "already exists")
}

func (s *cdcService) getNewArchTiCDCCR(
	ctx context.Context,
	proxy facade.ServerlessClusterProxyFacade,
	clusterName string,
) (*v1alpha1.TidbCluster, bizerr.BizError) {
	data, err := proxy.CallK8sResource(ctx, &clusterproxy.K8sResourceRequest{
		OperateType: "Get",
		APIVersion:  tiCDCResourceAPIVersion,
		Kind:        tiCDCResourceKind,
		Namespace:   config.Global().CDCConfig.NewArch.Namespace,
		Name:        clusterName,
	})
	if err != nil {
		var respErr *tidbcloud.ResponseError
		if errors.As(err, &respErr) && respErr.StatusCode == http.StatusNotFound {
			return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "ticdc cluster %s not found", clusterName)
		}
		return nil, bizerr.WrapError(err, "get ticdc cr failed")
	}
	cr, convErr := fromK8sResourceTiCDCCR(data)
	if convErr != nil {
		return nil, bizerr.WrapError(convErr, "decode ticdc cr failed")
	}
	return cr, nil
}

func (s *cdcService) createNewArchTiCDCCR(
	ctx context.Context,
	proxy facade.ServerlessClusterProxyFacade,
	cr *v1alpha1.TidbCluster,
) bizerr.BizError {
	obj, err := toK8sResourceObject(cr)
	if err != nil {
		return bizerr.WrapError(err, "encode ticdc cr failed")
	}
	_, err = proxy.CallK8sResource(ctx, &clusterproxy.K8sResourceRequest{
		OperateType: "Create",
		APIVersion:  tiCDCResourceAPIVersion,
		Kind:        tiCDCResourceKind,
		Namespace:   cr.Namespace,
		Object:      obj,
	})
	if err != nil {
		// Idempotency: a previous reconcile may have created the CR but
		// failed to record DB state, or two reconciles may race. Either
		// way the CR is now present, so treat 409 Conflict as success
		// and let the caller fall through to its update path.
		if isAlreadyExistsErr(err) {
			return nil
		}
		return bizerr.WrapError(err, "create ticdc cr failed")
	}
	return nil
}

func (s *cdcService) updateNewArchTiCDCCR(
	ctx context.Context,
	proxy facade.ServerlessClusterProxyFacade,
	cr *v1alpha1.TidbCluster,
) bizerr.BizError {
	obj, err := toK8sResourceObject(cr)
	if err != nil {
		return bizerr.WrapError(err, "encode ticdc cr failed")
	}
	_, err = proxy.CallK8sResource(ctx, &clusterproxy.K8sResourceRequest{
		OperateType: "Update",
		APIVersion:  tiCDCResourceAPIVersion,
		Kind:        tiCDCResourceKind,
		Namespace:   cr.Namespace,
		Name:        cr.Name,
		Object:      obj,
	})
	if err != nil {
		return bizerr.WrapError(err, "update ticdc cr failed")
	}
	return nil
}

// The shared TiCDC CR is updated by both serverless-service and
// tidb-operator. Re-read and retry when the API server rejects an Update
// because the resourceVersion became stale between Get and Update.
const newArchTiCDCCRUpdateMaxAttempts = 5

func isNewArchTiCDCCRConflictErr(err error) bool {
	if err == nil {
		return false
	}
	var respErr *tidbcloud.ResponseError
	if errors.As(err, &respErr) && respErr.StatusCode == http.StatusConflict {
		// A Create conflict means "already exists", not a stale Update.
		return !strings.Contains(strings.ToLower(respErr.Error()), "already exists")
	}
	msg := strings.ToLower(err.Error())
	if strings.Contains(msg, "already exists") {
		return false
	}
	return strings.Contains(msg, "the object has been modified") ||
		strings.Contains(msg, "please apply your changes to the latest version")
}

func (s *cdcService) applyNewArchTiCDCCRDesiredSpec(
	ctx context.Context,
	proxy facade.ServerlessClusterProxyFacade,
	clusterName string,
	existingCR, desired *v1alpha1.TidbCluster,
) bizerr.BizError {
	cr := existingCR
	for attempt := 0; attempt < newArchTiCDCCRUpdateMaxAttempts; attempt++ {
		if attempt > 0 {
			latest, err := s.getNewArchTiCDCCR(ctx, proxy, clusterName)
			if err != nil {
				return err
			}
			cr = latest
		}
		applyEssentialTiCDCCRDesiredSpec(cr, desired)
		err := s.updateNewArchTiCDCCR(ctx, proxy, cr)
		if err == nil {
			return nil
		}
		if !isNewArchTiCDCCRConflictErr(err) {
			return err
		}
	}
	return bizerr.New(
		bizerr.ErrCodeUnknown,
		"update ticdc cr %s still conflicting after %d attempts",
		clusterName,
		newArchTiCDCCRUpdateMaxAttempts,
	)
}

func (s *cdcService) deleteNewArchTiCDCCR(
	ctx context.Context,
	proxy facade.ServerlessClusterProxyFacade,
	clusterName string,
) bizerr.BizError {
	_, err := proxy.CallK8sResource(ctx, &clusterproxy.K8sResourceRequest{
		OperateType: "Delete",
		APIVersion:  tiCDCResourceAPIVersion,
		Kind:        tiCDCResourceKind,
		Namespace:   config.Global().CDCConfig.NewArch.Namespace,
		Name:        clusterName,
	})
	if err != nil {
		var respErr *tidbcloud.ResponseError
		if errors.As(err, &respErr) && respErr.StatusCode == http.StatusNotFound {
			return bizerr.New(bizerr.ErrCodeRecordNotFound, "ticdc cluster %s not found", clusterName)
		}
		return bizerr.WrapError(err, "delete ticdc cr failed")
	}
	return nil
}

// cleanupNewArchTiCDCDataplane removes the per-shard dataplane bits
// that ensureNewArchTiCDCCluster created alongside the TidbCluster CR:
//
//   - the 2 cert-manager Certificate CRs (cluster-client + ticdc-cluster),
//     which would otherwise outlive the TidbCluster as orphan certs;
//   - the StatefulSet's `ticdc-sort-kv-<cr>-ticdc-<ordinal>` PVCs.
//     tidb-operator removes the StatefulSet when the TidbCluster goes
//     away, but the PVCs survive (k8s StatefulSet semantics) and each
//     keeps a real EBS / disk volume billed to the customer.
//
// Caller must have already deleted the TidbCluster CR (so the operator
// has reaped the StatefulSet + pods). All deletions are NotFound-tolerant
// so the routine is safe to retry and safe to run on a shard that was
// only partially provisioned.
//
// NOT cleaned up here:
//   - StorageClass `cdc-{cloud}-{iops}-{tput}` is cluster-scoped and
//     shared across every shard on this dataplane — only an opt-in
//     dataplane teardown should remove it;
//   - cert-manager Issuer is operator-managed infrastructure.
//
// `maxPVCOrdinal` covers the largest Replicas value in the rcu ladder
// (currently 6 at 48rcu, with headroom for future tiers).
func (s *cdcService) cleanupNewArchTiCDCDataplane(
	ctx context.Context,
	proxy facade.ServerlessClusterProxyFacade,
	clusterName string,
) bizerr.BizError {
	cfg := config.Global().CDCConfig.NewArch

	// 1. Delete the two Certificate CRs by exact name.
	certNames := []string{
		clusterName + "-cluster-client-secret",
		clusterName + "-ticdc-cluster-secret",
	}
	for _, name := range certNames {
		_, err := proxy.CallK8sResource(ctx, &clusterproxy.K8sResourceRequest{
			OperateType: "Delete",
			APIVersion:  certManagerAPIVersion,
			Kind:        certManagerKind,
			Namespace:   cfg.Namespace,
			Name:        name,
		})
		if err != nil && !isNotFoundErr(err) {
			return bizerr.WrapError(err, "delete certificate %s failed", name)
		}
	}

	// 2. Delete the StatefulSet PVCs. tidb-operator names them
	// `ticdc-{volumeName}-{stsName}-{ordinal}`, where stsName is
	// `{clusterName}-ticdc` and volumeName is tiCDCDefaultDirName.
	const maxPVCOrdinal = 16
	pvcPrefix := fmt.Sprintf("ticdc-%s-%s-ticdc-", tiCDCDefaultDirName, clusterName)
	for i := 0; i < maxPVCOrdinal; i++ {
		name := fmt.Sprintf("%s%d", pvcPrefix, i)
		_, err := proxy.CallK8sResource(ctx, &clusterproxy.K8sResourceRequest{
			OperateType: "Delete",
			APIVersion:  "v1",
			Kind:        "PersistentVolumeClaim",
			Namespace:   cfg.Namespace,
			Name:        name,
		})
		if err == nil {
			continue
		}
		if isNotFoundErr(err) {
			// First NotFound means we've gone past the highest
			// ordinal that ever existed (or the PVCs are already
			// gone). Either way: stop probing further ordinals.
			return nil
		}
		return bizerr.WrapError(err, "delete pvc %s failed", name)
	}
	return nil
}

// isNotFoundErr unwraps clusterproxy responses to detect HTTP 404.
// Mirrors getNewArchTiCDCCR's inline check; pulled out so the cleanup
// + delete paths share one helper.
func isNotFoundErr(err error) bool {
	if err == nil {
		return false
	}
	var respErr *tidbcloud.ResponseError
	if errors.As(err, &respErr) && respErr.StatusCode == http.StatusNotFound {
		return true
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "not found")
}

func (s *cdcService) ensureNewArchTiCDCCluster(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
) (string, bizerr.BizError) {
	shard, err := s.ensureNewArchChangefeedResource(ctx, changefeed, formatNewArchSpecification(changefeed.RCU))
	if err != nil {
		return "", err
	}
	proxy, err := s.getServerlessClusterProxy(ctx, changefeed.CloudProvider, changefeed.Region, changefeed.Kubernetes)
	if err != nil {
		return "", err
	}
	resources, err := s.listNewArchChangefeedResourcesForShard(ctx, shard)
	if err != nil {
		return "", err
	}
	// Resolve version up front: existing shard sticks to its locked-in
	// version; a brand-new shard adopts the configured default.
	version := s.resolveShardVersion(ctx, shard)
	activeResources := filterExpectedNewArchResources(resources)
	if len(activeResources) == 0 {
		if err := s.saveNewArchTiCDCCluster(
			ctx,
			shard,
			"",
			"",
			version,
			constant.TiCDCClusterStateDeleted,
		); err != nil {
			return "", err
		}
		return shard.ClusterName, nil
	}
	totalRCU, parseErr := sumNewArchResourceRCU(activeResources)
	if parseErr != nil {
		return "", bizerr.WrapError(parseErr, "sum newarch resource rcu failed")
	}
	instType, replicas := selectNewArchTiCDCInstance(totalRCU)
	specName := formatNewArchSpecification(totalRCU)
	if err := s.ensureNewArchTiCDCCerts(ctx, proxy, shard.ClusterName); err != nil {
		return "", err
	}
	scName, err := s.ensureNewArchStorageClass(ctx, proxy, changefeed.CloudProvider, instType)
	if err != nil {
		return "", err
	}
	existingCR, getErr := s.getNewArchTiCDCCR(ctx, proxy, shard.ClusterName)
	if getErr != nil {
		if getErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			desired := buildEssentialTiCDCCR(
				shard.ClusterName,
				changefeed.CloudProvider,
				changefeed.Region,
				instType,
				replicas,
				scName,
				version,
			)
			if err := s.createNewArchTiCDCCR(ctx, proxy, desired); err != nil {
				return "", err
			}
			if err := s.saveNewArchTiCDCCluster(
				ctx,
				shard,
				specName,
				"",
				version,
				constant.TiCDCClusterStateCreating,
			); err != nil {
				return "", err
			}
			return shard.ClusterName, nil
		}
		return "", getErr
	}
	desired := buildEssentialTiCDCCR(
		shard.ClusterName,
		changefeed.CloudProvider,
		changefeed.Region,
		instType,
		replicas,
		scName,
		version,
	)
	if err := s.applyNewArchTiCDCCRDesiredSpec(ctx, proxy, shard.ClusterName, existingCR, desired); err != nil {
		return "", err
	}
	if err := s.saveNewArchTiCDCCluster(
		ctx,
		shard,
		specName,
		"",
		version,
		constant.TiCDCClusterStateCreating,
	); err != nil {
		return "", err
	}
	return shard.ClusterName, nil
}

func (s *cdcService) reconcileNewArchTiCDCClusterForShard(
	ctx context.Context,
	shard *newArchShard,
) bizerr.BizError {
	resources, err := s.listNewArchChangefeedResourcesForShard(ctx, shard)
	if err != nil {
		return err
	}
	version := s.resolveShardVersion(ctx, shard)
	proxy, err := s.getServerlessClusterProxy(ctx, shard.Provider, shard.Region, shard.Kubernetes)
	if err != nil {
		return err
	}
	activeResources := filterExpectedNewArchResources(resources)
	if len(activeResources) == 0 {
		// Delete the TidbCluster first (tidb-operator reaps the
		// StatefulSet + pods), then clean up the per-shard cert + PVCs
		// that the operator leaves behind. Cleanup is best-effort but
		// any non-NotFound error is fatal so the shard stays in the
		// "needs cleanup" state until the next reconcile retries.
		delErr := s.deleteNewArchTiCDCCR(ctx, proxy, shard.ClusterName)
		if delErr != nil && delErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return delErr
		}
		if cleanupErr := s.cleanupNewArchTiCDCDataplane(ctx, proxy, shard.ClusterName); cleanupErr != nil {
			return cleanupErr
		}
		return s.markNewArchTiCDCClusterDeleted(
			ctx,
			shard,
			version,
			resources,
		)
	}
	totalRCU, parseErr := sumNewArchResourceRCU(activeResources)
	if parseErr != nil {
		return bizerr.WrapError(parseErr, "sum newarch resource rcu failed")
	}
	instType, replicas := selectNewArchTiCDCInstance(totalRCU)
	specName := formatNewArchSpecification(totalRCU)
	if err := s.ensureNewArchTiCDCCerts(ctx, proxy, shard.ClusterName); err != nil {
		return err
	}
	scName, err := s.ensureNewArchStorageClass(ctx, proxy, shard.Provider, instType)
	if err != nil {
		return err
	}
	existingCR, getErr := s.getNewArchTiCDCCR(ctx, proxy, shard.ClusterName)
	if getErr != nil {
		if getErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			desired := buildEssentialTiCDCCR(
				shard.ClusterName,
				shard.Provider,
				shard.Region,
				instType,
				replicas,
				scName,
				version,
			)
			if err := s.createNewArchTiCDCCR(ctx, proxy, desired); err != nil {
				return err
			}
			return s.saveNewArchTiCDCCluster(
				ctx,
				shard,
				specName,
				"",
				version,
				constant.TiCDCClusterStateCreating,
			)
		}
		return getErr
	}
	desired := buildEssentialTiCDCCR(
		shard.ClusterName,
		shard.Provider,
		shard.Region,
		instType,
		replicas,
		scName,
		version,
	)
	if err := s.applyNewArchTiCDCCRDesiredSpec(ctx, proxy, shard.ClusterName, existingCR, desired); err != nil {
		return err
	}
	return s.saveNewArchTiCDCCluster(
		ctx,
		shard,
		specName,
		"",
		version,
		constant.TiCDCClusterStateCreating,
	)
}

func (s *cdcService) syncNewArchChangefeedResourcesForShard(
	ctx context.Context,
	resources []*entity.TiCDCChangefeedResource,
) error {
	if len(resources) == 0 {
		return nil
	}
	shard := shardFromResource(resources[0])
	allResources, err := s.listNewArchChangefeedResourcesForShard(ctx, &shard)
	if err != nil {
		return err
	}

	proxy, err := s.getServerlessClusterProxy(ctx, shard.Provider, shard.Region, shard.Kubernetes)
	if err != nil {
		return err
	}
	activeResources := filterExpectedNewArchResources(allResources)
	// dataplane GET runs OUTSIDE the DB transaction: it can be slow
	// and we don't want to hold a row lock across a k8s API call.
	cr, bizErr := s.getNewArchTiCDCCR(ctx, proxy, shard.ClusterName)
	crFound := bizErr == nil
	if bizErr != nil && bizErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return bizErr
	}
	version := s.resolveShardVersion(ctx, &shard)

	if !crFound {
		return s.syncOneDeletedNewArchTiCDCCluster(
			ctx,
			&shard,
			version,
			activeResources,
			allResources,
		)
	}

	return s.syncOneExistingNewArchTiCDCCluster(
		ctx,
		&shard,
		version,
		proxy,
		cr,
		activeResources,
		allResources,
	)
}

func (s *cdcService) syncOneDeletedNewArchTiCDCCluster(
	ctx context.Context,
	shard *newArchShard,
	version string,
	activeResources, allResources []*entity.TiCDCChangefeedResource,
) error {
	// Everything below is DB-only and benefits from being one atomic
	// write: the per-shard ticdc_cluster row and the per-changefeed
	// resource rows must agree about which spec the shard is at,
	// otherwise a partial failure (cluster row says running, but a
	// resource row still has actual='') leaves readers confused.
	return s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		// CR is gone from dataplane.
		if len(activeResources) != 0 {
			// Someone still wants this shard alive - flag creating
			// so the next ensure/reconcile pass will rebuild the CR.
			totalRCU, parseErr := sumNewArchResourceRCU(activeResources)
			if parseErr != nil {
				return parseErr
			}
			specName := formatNewArchSpecification(totalRCU)
			return s.saveNewArchTiCDCCluster(
				ctx, shard,
				specName, "", version, constant.TiCDCClusterStateCreating,
			)
		}
		// Nobody wants this shard - mark deleted and ack every
		// resource's actual_specification to its expected state.
		return s.markNewArchTiCDCClusterDeleted(ctx, shard, version, allResources)
	})
}

func (s *cdcService) syncOneExistingNewArchTiCDCCluster(
	ctx context.Context,
	shard *newArchShard,
	version string,
	proxy facade.ServerlessClusterProxyFacade,
	cr *v1alpha1.TidbCluster,
	activeResources, allResources []*entity.TiCDCChangefeedResource,
) error {
	if len(activeResources) == 0 {
		return s.syncNewArchTiCDCClusterDeleted(
			ctx,
			shard,
			version,
			proxy,
			allResources,
		)
	}
	return s.syncNewArchTiCDCClusterRunning(
		ctx,
		shard,
		version,
		cr,
		activeResources,
		allResources,
	)
}

func (s *cdcService) syncNewArchTiCDCClusterDeleted(
	ctx context.Context,
	shard *newArchShard,
	version string,
	proxy facade.ServerlessClusterProxyFacade,
	allResources []*entity.TiCDCChangefeedResource,
) error {
	// The CR still exists (or is deleting), but the desired shard spec
	// is empty. Keep driving dataplane cleanup outside the DB
	// transaction, then release actual_specification below.
	delErr := s.deleteNewArchTiCDCCR(ctx, proxy, shard.ClusterName)
	if delErr != nil && delErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return delErr
	}
	if cleanupErr := s.cleanupNewArchTiCDCDataplane(ctx, proxy, shard.ClusterName); cleanupErr != nil {
		return cleanupErr
	}

	return s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		// Dataplane cleanup already happened outside the transaction.
		// Now ack actual_specification to expected, matching
		// dataflow-ng's markTiCDCClusterDeleted semantics.
		return s.markNewArchTiCDCClusterDeleted(ctx, shard, version, allResources)
	})
}

func (s *cdcService) syncNewArchTiCDCClusterRunning(
	ctx context.Context,
	shard *newArchShard,
	version string,
	cr *v1alpha1.TidbCluster,
	activeResources, allResources []*entity.TiCDCChangefeedResource,
) error {
	return s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		totalRCU, parseErr := sumNewArchResourceRCU(activeResources)
		if parseErr != nil {
			return parseErr
		}
		instType, replicas := selectNewArchTiCDCInstance(totalRCU)
		specName := formatNewArchSpecification(totalRCU)
		if !isTiCDCCRReady(cr) || !isTiCDCCRSpecMatched(cr, instType, replicas) {
			// CR is there but not yet at the target spec - leave the
			// shard in creating, don't promote resource rows to actual.
			return s.saveNewArchTiCDCCluster(
				ctx, shard,
				specName, "", version, constant.TiCDCClusterStateCreating,
			)
		}

		// CR is ready AND matches expected spec - promote the shard
		// to running and ack each resource row's expected==actual.
		return s.markNewArchTiCDCClusterRunning(
			ctx,
			shard,
			specName,
			version,
			allResources,
		)
	})
}