package dedicatedregional

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"path"
	"regexp"
	"strconv"
	"strings"

	gcpcompute "cloud.google.com/go/compute/apiv1"
	gcpcomputepb "cloud.google.com/go/compute/apiv1/computepb"
	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	"github.com/crossplane-contrib/provider-aws/apis/eks/manualv1alpha1"
	eksv1beta1 "github.com/crossplane-contrib/provider-aws/apis/eks/v1beta1"
	awsiamv1alpha1 "github.com/crossplane-contrib/provider-aws/apis/iam/v1alpha1"
	awsiamv1beta1 "github.com/crossplane-contrib/provider-aws/apis/iam/v1beta1"
	route53v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/route53/v1alpha1"
	computev1alpha1 "github.com/crossplane-contrib/provider-gcp/apis/compute/v1alpha1"
	computev1beta1 "github.com/crossplane-contrib/provider-gcp/apis/compute/v1beta1"

	ec2v1beta1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1beta1"
	awsv1beta1 "github.com/crossplane-contrib/provider-aws/apis/v1beta1"
	gkev1beta1 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	gkev1beta2 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta2"
	gcpiamv1alpha1 "github.com/crossplane-contrib/provider-gcp/apis/iam/v1alpha1"
	gcpv1beta1 "github.com/crossplane-contrib/provider-gcp/apis/v1beta1"
	"github.com/crossplane/crossplane-runtime/pkg/meta"
	xpmeta "github.com/crossplane/crossplane-runtime/pkg/meta"
	vpcpeeringv1beta1 "github.com/crossplane/provider-gcp/apis/vpcpeering/v1beta1"

	"github.com/go-logr/logr"
	"github.com/pingcap/log"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/util/retry"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"

	regionalpb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"

	cloudeksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	clusterctlpb "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	infranamer "github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/clustercli"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/constants"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/models"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	rsctl "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	appboxtask "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/appbox/task"
	appboxekstask "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/appbox/task/eks"
	appboxgketask "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/appbox/task/gke"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/account"
)

var (
	patchAnnotationTPL    = `{ "metadata": {"annotations": {"%s": "%s"}}}`
	patchAnnotationMapTPL = `{ "metadata": {"annotations": %s }}`
)

// var patchAPPBoxSpecPausedTPL = `{"spec": {"paused": %v}}`
var (
	patchAPPBoxStatusProxyBaseURLTPL     = `{"status": {"proxyBaseUrl": "%s"}}`
	patchAPPBoxStatusDataflowAgentURLTPL = `{"status": {"dataflowAgentUrl": "%s"}}`
)

var (
	roleArnRE = regexp.MustCompile(`^arn:aws:iam::(\d+):role\/([a-zA-Z0-9-]+)`)
	//nolint:unused
	addonVersionRE = regexp.MustCompile(`v\d+\.\d+\.\d+-eksbuild\.\d+`)
)

type Option func(*Service) error

type Service struct {
	// we have some special case when creating VPCPeering CR, need to read kube-apiserver directly.
	kubeClient    client.Client // access kube-apiserver with mgr cache.
	kubeAPIReader client.Reader // access kube-apiserver directly.

	provider      string // value can be `eks` or `gke`
	region        string // phyical region, value can be `us-west-2`, `us-east-2`, `us-west1`
	config        rsconfig.RootConfig
	tmplconfig    rsconfig.TemplatesConfig
	accountR      account.AccountReader
	clusterClient clustercli.ClusterClient
	marshalOpt    protojson.MarshalOptions
	unmarshalOpt  protojson.UnmarshalOptions
	manager       *rsctl.Manager
	// client for gcp
	// these two client are only set on gcp region.
	subnetworkClient        *gcpcompute.SubnetworksClient
	serviceattachmentClient *gcpcompute.ServiceAttachmentsClient
}

func NewService(
	kubeClient client.Client,
	kubeAPIReader client.Reader,
	clusterClient clustercli.ClusterClient,
	config rsconfig.RootConfig,
	tmplconfig rsconfig.TemplatesConfig,
	rsmanager *rsctl.Manager,
	opts ...Option,
) (*Service, error) {
	accountR := account.NewAccountReader(kubeClient)
	svc := &Service{
		kubeClient:    kubeClient,
		kubeAPIReader: kubeAPIReader,
		config:        config,
		tmplconfig:    tmplconfig,
		accountR:      accountR,
		provider:      config.Provider,
		region:        config.RegionConfig.Name,
		clusterClient: clusterClient,
		manager:       rsmanager,
		marshalOpt: protojson.MarshalOptions{
			UseProtoNames:   false,
			EmitUnpopulated: true,
		},
		unmarshalOpt: protojson.UnmarshalOptions{
			DiscardUnknown: true,
		},
	}
	for _, opt := range opts {
		if err := opt(svc); err != nil {
			return nil, err
		}
	}
	return svc, nil
}

func (svc *Service) WithGCPSubnentworkClient(cli *gcpcompute.SubnetworksClient) *Service {
	svc.subnetworkClient = cli
	return svc
}

func (svc *Service) WithGCPServiceAttchementClient(cli *gcpcompute.ServiceAttachmentsClient) *Service {
	svc.serviceattachmentClient = cli
	return svc
}

type PObject[T any] interface {
	client.Object
	*T
}

type DeepCopyable[PT PObject[T], T any] interface {
	DeepCopy() PT
}

// we need a type which:
// 1. implement client.Object interface.
// 2. has a DeepCopy() method which return a copy of itself.
type DeepCopyableObject[PT PObject[T], T any] interface {
	DeepCopyable[PT, T]
	PObject[T]
}

func upsertObject[DT DeepCopyableObject[PT, T], PT PObject[T], T any](ctx context.Context, kubeClient client.Client, kubeAPIReader client.Reader, cr DT) error {
	key := client.ObjectKeyFromObject(cr).String()
	cp := cr.DeepCopy()
	if err := kubeClient.Create(ctx, cr); err != nil {
		log.Error("create object failed", zap.String("key", key), zap.Error(err))
		if !apierrors.IsAlreadyExists(err) {
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
		// overwrite the object if existed
		var existed PT = new(T)
		if err := kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(cr), existed); err != nil {
			log.Error("get object failed", zap.Error(err), zap.String("key", key))
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
		if cr.GetAnnotations()[v1alpha1.AnnotationKeyReconciliationPaused] != "true" {
			return bizerr.New(bizerr.ResourceConflict, "resource is not paused")
		}
		// commonly, we may implement the update like following:
		// existed.Annotations =  cr.Annotaions
		// existed.Labels = cr.Labels
		// existed.Spec = cr.Spec
		// existed.Status = cr.Status
		// _ := kubeClient.Updated(ctx, cr)
		// however, we cannot access .Spec and .Status field directly since we use generics here
		// so we set resourceversion and uid on cr and update cr as a workaround
		// set these fields to solve precondition error
		cr.SetResourceVersion(existed.GetResourceVersion())
		cr.SetUID(existed.GetUID())
		cp := cr.DeepCopy()
		if err := kubeClient.Update(ctx, cr); err != nil {
			log.Error("update object failed", zap.String("key", key), zap.Error(err))
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
		// set these fields to solve conflict error
		err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
			if err := kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(existed), existed); err != nil {
				log.Error("get object failed", zap.Error(err), zap.String("key", key))
				return err
			}
			// set these fields to solve conflict error
			cp.SetResourceVersion(existed.GetResourceVersion())
			cp.SetUID(existed.GetUID())
			if err := kubeClient.Status().Update(ctx, cp); err != nil {
				log.Error("update status failed",
					zap.String("key", key),
					zap.String("resourceversion", cp.GetResourceVersion()),
					zap.Int64("generation", cp.GetGeneration()),
					zap.Error(err))
				return err
			}
			return nil
		})
		if err != nil {
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	} else {
		err := retry.OnError(retry.DefaultRetry,
			func(err error) bool {
				return apierrors.IsConflict(err) || apierrors.IsNotFound(err)
			},
			func() error {
				var existed PT = new(T)
				if err := kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(cr), existed); err != nil {
					log.Error("get object failed", zap.String("key", key), zap.Error(err))
					return err
				}
				// still need to update status with the given cr for the first creation
				cp.SetResourceVersion(existed.GetResourceVersion())
				cp.SetUID(existed.GetUID())
				if err := kubeClient.Status().Update(ctx, cp); err != nil {
					log.Error("update status failed", zap.String("key", key),
						zap.String("resourceversion", cp.GetResourceVersion()),
						zap.Int64("generation", cp.GetGeneration()),
						zap.Error(err))
					return err
				}
				return nil
			})
		if err != nil {
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}
	return nil
}

func (svc *Service) checkProviderConfigExistedAWS(ctx context.Context, name string) (*awsv1beta1.ProviderConfig, error) {
	providerConfig := new(awsv1beta1.ProviderConfig)
	providerConfig.Name = name
	if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(providerConfig), providerConfig); err != nil {
		log.Error("get providerconfig failed", zap.Any("key", client.ObjectKeyFromObject(providerConfig)), zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, err.Error())
		}
		return nil, bizerr.New(bizerr.K8SClientError, err.Error())
	}
	return providerConfig, nil
}

func (svc *Service) checkProviderConfigExistedGCP(ctx context.Context, name string) (*gcpv1beta1.ProviderConfig, error) {
	providerConfig := new(gcpv1beta1.ProviderConfig)
	providerConfig.Name = name
	if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(providerConfig), providerConfig); err != nil {
		log.Error("get providerconfig failed", zap.Any("key", client.ObjectKeyFromObject(providerConfig)), zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, err.Error())
		}
		return nil, bizerr.New(bizerr.K8SClientError, err.Error())
	}
	return providerConfig, nil
}

func (svc *Service) ImportAPPBox(ctx context.Context, cr *v1alpha1.APPBox) error {
	if !cr.DeletionTimestamp.IsZero() {
		return bizerr.New(bizerr.ParameterInvalid, "deletiontimestamp has value")
	}
	// check label and annotation
	var tenantID, projectID string
	for key, value := range cr.Labels {
		if strings.HasPrefix(key, v1alpha1.TenantIDKeyPrefix) && value == "true" {
			tenantID = strings.TrimPrefix(key, v1alpha1.TenantIDKeyPrefix)
		}
		if strings.HasPrefix(key, v1alpha1.ProjectIDKeyPrefix) && value == "true" {
			projectID = strings.TrimPrefix(key, v1alpha1.ProjectIDKeyPrefix)
		}
	}
	if tenantID == "" {
		return bizerr.New(bizerr.ParameterInvalid, "tenantId label is empty")
	}
	if cr.Labels[v1alpha1.TagTenantKey] != tenantID {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is mismatched", v1alpha1.TagTenantKey))
	}
	if projectID == "" {
		return bizerr.New(bizerr.ParameterInvalid, "projectId label is empty")
	}
	if cr.Labels[v1alpha1.TagProjectKey] != projectID {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is mismatched", v1alpha1.TagProjectKey))
	}
	if cr.Labels[v1alpha1.VPCNameKey] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is empty", v1alpha1.VPCNameKey))
	}
	if cr.Spec.KubernetesConfig.Version == "" {
		return bizerr.New(bizerr.ParameterInvalid, "Spec.KubernetesConfig.Version is empty")
	}
	if cr.Spec.VPCRef.Name == "" {
		return bizerr.New(bizerr.ParameterInvalid, "Spec.VPCRef.Name is empty")
	}
	if cr.Labels[v1alpha1.VPCNameKey] != cr.Spec.VPCRef.Name {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is mismatched", v1alpha1.VPCNameKey))
	}
	// this anno is used to overwrite the providerconfig name in eks gke cr spec.
	if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameProvidreConfig] == "" {
		return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-providerconfig anno is empty")
	}
	switch svc.provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameEKS] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-eks anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameIAMRole] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-iamrole anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameInstanceProfile] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-instanceprofile anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameNodeGroup] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-nodegroup anno is empty")
		}
		if cr.Status.IAM.NodeRoleARN == "" {
			return bizerr.New(bizerr.ParameterInvalid, "Status.IAM.NodeRoleARN is empty")
		}
		if cr.Spec.Provider != "eks" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Provider is mismatched")
		}
	case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameServiceAccount] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-serviceaccount anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameServiceAccountKey] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-serviceaccountkey anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameServiceAccountPolicy] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-serviceaccountpolicy anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameGKE] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-gke anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameNodePool] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-nodepool anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameOIDCProvider] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-oidcprovider anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameIAMRolePCAIssuer] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-iamrole-pcaissuer anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameIAMRoleSLIReporter] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-iamrole-slireporter anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnSpecMaxPodsPerNode] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-spec-maxpodspernode anno is empty")
		}
		if cr.Spec.Provider != "gke" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Provider is mismatched")
		}
		if cr.Status.IAM.ServiceAccountEmail == "" {
			return bizerr.New(bizerr.ParameterInvalid, "Status.IAM.ServiceAccountEmail is empty")
		}
	}
	// set fields
	cr.Labels[v1alpha1.IsLegacyKey] = "true"
	cr.Labels[v1alpha1.RegionKey] = svc.region

	switch svc.provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		cr.Labels = util.MergeMaps(cr.Labels, svc.config.ResourceLabels)
	case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
		// tag label will be populate to gke resource labels
		// add resource labels is not a safe operation and may update nodes
		// so we skip merging additional resource labels fo migrated gke
	}
	if cr.Annotations == nil {
		cr.Annotations = map[string]string{}
	}
	// must set paused anno to true
	cr.Annotations[v1alpha1.AnnotationKeyReconciliationPaused] = "true"
	switch svc.provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		cr.Annotations[v1alpha1.AnnotationKeyLegacyAPPBoxSkipEKSTask] = "true"
	case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
		cr.Annotations[v1alpha1.AnnotationKeyLegacyAPPBoxSkipGKETask] = "true"
	}

	cr.SetFinalizers([]string{appboxtask.Finalizer})
	namer := utils.NewNamer(tenantID, projectID, "", svc.config.RegionConfig.Name, svc.config.Env)
	cr.Namespace = namer.GetNamespace()
	// VPCRef.Namespace is set by this handler
	// VPCRef.Name is set by caller
	cr.Spec.VPCRef.Namespace = namer.GetNamespace()
	if templateID := svc.tmplconfig.GetDefaultAddonTemplateId(); templateID == "" {
		return bizerr.New(bizerr.ConfigInvalid, "cannot find addonTemplateId in config")
	} else {
		cr.Spec.KubernetesConfig.AddonTemplateID = templateID
	}
	if templateID := svc.tmplconfig.GetDefaultInfraTemplateIdWithVersion(cr.Spec.KubernetesConfig.Version); templateID == "" {
		return bizerr.New(bizerr.ConfigInvalid, fmt.Sprintf("cannot find infraTemplateId for k8s version %s in config", cr.Spec.KubernetesConfig.Version))
	} else {
		cr.Spec.KubernetesConfig.InfraTemplateID = templateID
	}
	if templateID := svc.tmplconfig.GetDefaultDedicatedTemplateIdWithVersion(cr.Spec.KubernetesConfig.Version); templateID == "" {
		return bizerr.New(bizerr.ConfigInvalid, fmt.Sprintf("cannot find dedicatedTemplateId for k8s version %s in config", cr.Spec.KubernetesConfig.Version))
	} else {
		cr.Spec.KubernetesConfig.DedicatedTemplateID = templateID
	}
	if templateID := svc.tmplconfig.GetDefaultDPTemplateIdWithVersion(cr.Spec.KubernetesConfig.Version); templateID == "" {
		return bizerr.New(bizerr.ConfigInvalid, fmt.Sprintf("cannot find dpTemplateId for k8s version %s in config", cr.Spec.KubernetesConfig.Version))
	} else {
		cr.Spec.KubernetesConfig.DPTemplateID = templateID
	}
	cr.Status.Phase = v1alpha1.APPBoxPhaseActive
	cr.Status.Reason = "migrated from infra-apiserver"

	if err := upsertObject[*v1alpha1.APPBox](ctx, svc.kubeClient, svc.kubeAPIReader, cr); err != nil {
		log.Error("update appbox failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
		return err
	}
	return nil
}

func (svc *Service) CheckAPPBox(ctx context.Context, key client.ObjectKey) error {
	cr := new(v1alpha1.APPBox)
	namer := utils.NewNamer("", "", "", svc.config.RegionConfig.Name, svc.config.Env)
	key.Namespace = namer.GetNamespace()
	if err := svc.kubeAPIReader.Get(ctx, key, cr); err != nil {
		if apierrors.IsNotFound(err) {
			return bizerr.New(bizerr.ResourceNotFound, err.Error())
		}
		log.Error("get appbox failed", zap.String("key", key.String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}

	isHibernated := cr.Annotations[v1alpha1.AnnotationKeyIsHiberated] == "true"
	// check role, policy, rolepolicyattachment existed and Status.IAM.NodeRoleARN
	// check instanceprofile existed
	// check eks/gke cluster existed and Status.KubernetesClusterName
	// check nodegroup, launchtemplate existed
	// set Status.ProxyBaseUrl
	// only need to check k8s version is set
	vpc := new(v1alpha1.VPC)
	if ref := cr.Spec.VPCRef; ref != nil && ref.Name != "" {
		vpc.Namespace = cr.Namespace
		vpc.Name = ref.Name
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(vpc), vpc); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			log.Error("get rs vpc failed", zap.String("key", client.ObjectKeyFromObject(vpc).String()), zap.Error(err))
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}
	// o11yroute for control plane has the same name as appbox
	// o11yroute not existed for hibernated case
	if !isHibernated {
		o11yroute := new(v1alpha1.O11yRoute)
		o11yroute.Name = cr.Name
		o11yroute.Namespace = cr.Namespace
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(o11yroute), o11yroute); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			log.Error("get o11yroute failed", zap.String("key", client.ObjectKeyFromObject(o11yroute).String()), zap.Error(err))
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}
	switch svc.provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameEKS]; name != "" {
			ekscluster := new(eksv1beta1.Cluster)
			ekscluster.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(ekscluster), ekscluster); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get eks cluster failed", zap.String("key", client.ObjectKeyFromObject(ekscluster).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if cr.Status.KubernetesClusterName != ekscluster.Name {
				return bizerr.New(bizerr.K8SClientError, "Status.KubernetesClusterName is mismatched")
			}
			taskCtx, err := (&appboxtask.Context{
				Config: svc.config,
				Local:  svc.kubeClient,
				VPC:    vpc.DeepCopy(),
				APPBox: cr.DeepCopy(),
				Logger: logr.FromSlogHandler(slog.Default().Handler()),
			}).Default()
			if err != nil {
				return bizerr.New(bizerr.InternalServerError, err.Error())
			}
			eksTask := appboxekstask.NewEKS(taskCtx, nil)
			ok, err := eksTask.IsSafeToApplyEKS(ctx, ekscluster)
			if err != nil {
				log.Error("check IsSafeToApplyEKS failed", zap.Error(err))
				return bizerr.New(bizerr.InternalServerError, err.Error())
			}
			if !ok {
				return bizerr.New(bizerr.ParameterInvalid, "not safe to apply eks for legacy appbox")
			}
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameIAMRole]; name != "" {
			iamrole := new(awsiamv1beta1.Role)
			iamrole.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(iamrole), iamrole); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get iamrole failed", zap.String("key", client.ObjectKeyFromObject(iamrole).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if cr.Status.IAM.NodeRoleARN != iamrole.Status.AtProvider.ARN {
				return bizerr.New(bizerr.K8SClientError, "Status.IAM.NodeRoleARN is mismatched")
			}
			matchLabels := client.MatchingLabels{appboxekstask.RoleRefLabelKey: name}
			{
				list := new(awsiamv1beta1.RolePolicyAttachmentList)
				if err := svc.kubeAPIReader.List(ctx, list, matchLabels); err != nil {
					log.Error("list rolepolicyattachment failed", zap.Error(err))
					return bizerr.New(bizerr.K8SClientError, err.Error())
				}
				if len(list.Items) == 0 {
					return bizerr.New(bizerr.ResourceNotFound, "rolepolicyattchment not found")
				}
			}
			{
				list := new(awsiamv1beta1.PolicyList)
				if err := svc.kubeAPIReader.List(ctx, list, matchLabels); err != nil {
					log.Error("list policy failed", zap.Error(err))
					return bizerr.New(bizerr.K8SClientError, err.Error())
				}
				if len(list.Items) == 0 {
					return bizerr.New(bizerr.ResourceNotFound, "policy not found")
				}

			}
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameInstanceProfile]; name != "" {
			instanceprofile := new(awsiamv1alpha1.InstanceProfile)
			instanceprofile.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(instanceprofile), instanceprofile); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameNodeGroup]; name != "" {
			nodegroup := new(manualv1alpha1.NodeGroup)
			nodegroup.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(nodegroup), nodegroup); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get nodegroup failed", zap.String("key", client.ObjectKeyFromObject(nodegroup).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			launchtemplate := new(ec2v1alpha1.LaunchTemplate)
			launchtemplate.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(launchtemplate), launchtemplate); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get launchtemplate failed", zap.String("key", client.ObjectKeyFromObject(launchtemplate).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
		}

	case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameServiceAccount]; name != "" {
			sa := new(gcpiamv1alpha1.ServiceAccount)
			sa.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(sa), sa); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get serviceaccount failed", zap.String("key", client.ObjectKeyFromObject(sa).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if cr.Status.IAM.ServiceAccountEmail != sa.Status.AtProvider.Email {
				return bizerr.New(bizerr.K8SClientError, "Status.IAM.ServiceAccountEmail is mismatched")
			}
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameServiceAccountKey]; name != "" {
			sakey := new(gcpiamv1alpha1.ServiceAccountKey)
			sakey.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(sakey), sakey); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get serviceaccountkey failed", zap.String("key", client.ObjectKeyFromObject(sakey).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameServiceAccountPolicy]; name == "" {
			sapolicy := new(gcpiamv1alpha1.ServiceAccountPolicy)
			sapolicy.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(sapolicy), sapolicy); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get serviceaccountpolicy failed", zap.String("key", client.ObjectKeyFromObject(sapolicy).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameGKE]; name != "" {
			gkecluster := new(gkev1beta2.Cluster)
			gkecluster.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(gkecluster), gkecluster); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get gkecluster failed", zap.String("key", client.ObjectKeyFromObject(gkecluster).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			taskCtx, err := (&appboxtask.Context{
				Config: svc.config,
				VPC:    vpc.DeepCopy(),
				APPBox: cr.DeepCopy(),
				Logger: logr.FromSlogHandler(slog.Default().Handler()),
			}).Default()
			if err != nil {
				return bizerr.New(bizerr.InternalServerError, err.Error())
			}
			gkeTask := appboxgketask.NewGKE(taskCtx, nil)
			gkeTask.Local = svc.kubeClient
			ok, err := gkeTask.IsSafeToApplyGKE(ctx, gkecluster)
			if err != nil {
				log.Error("IsSafeToApplyGKE failed", zap.String("key", client.ObjectKeyFromObject(gkecluster).String()), zap.Error(err))
				return bizerr.New(bizerr.InternalServerError, err.Error())
			}
			if !ok {
				return bizerr.New(bizerr.ParameterInvalid, "not safe to apply gke for legacy appbox")
			}
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameNodePool]; name != "" {
			nodepool := new(gkev1beta1.NodePool)
			nodepool.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(nodepool), nodepool); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get nodepool failed", zap.String("key", client.ObjectKeyFromObject(nodepool).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameOIDCProvider]; name != "" {
			oidcprovider := new(awsiamv1beta1.OpenIDConnectProvider)
			oidcprovider.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(oidcprovider), oidcprovider); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get oidcprovider failed", zap.String("key", client.ObjectKeyFromObject(oidcprovider).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameIAMRolePCAIssuer]; name != "" {
			iamrole := new(awsiamv1beta1.Role)
			iamrole.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(iamrole), iamrole); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get role for pcaissuer failed", zap.String("key", client.ObjectKeyFromObject(iamrole).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			iampolicy := new(awsiamv1beta1.Policy)
			iampolicy.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(iampolicy), iampolicy); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get policy for pcaissuer failed", zap.String("key", client.ObjectKeyFromObject(iampolicy).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			attachment := new(awsiamv1beta1.RolePolicyAttachment)
			attachment.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(attachment), attachment); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get rolepolicyattachment for pcaissuer failed", zap.String("key", client.ObjectKeyFromObject(attachment).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameIAMRoleSLIReporter]; name != "" {
			iamrole := new(awsiamv1beta1.Role)
			iamrole.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(iamrole), iamrole); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get role for slireporter failed", zap.String("key", client.ObjectKeyFromObject(iamrole).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			iampolicy := new(awsiamv1beta1.Policy)
			iampolicy.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(iampolicy), iampolicy); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get policy slireporter failed", zap.String("key", client.ObjectKeyFromObject(iampolicy).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			attachment := new(awsiamv1beta1.RolePolicyAttachment)
			attachment.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(attachment), attachment); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get rolepolicyattachment slireporter failed", zap.String("key", client.ObjectKeyFromObject(attachment).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
		}
	}
	// finally, we set the proxyBaseUrl, dataFlowAgentUrl and addonVersion
	// the related svc not existed under hibernated case
	if cr.Status.ProxyBaseUrl == "" && !isHibernated {
		addr, err := svc.getProxyBaseProxyURL(ctx, cr)
		if err != nil {
			return err
		}
		data := fmt.Sprintf(patchAPPBoxStatusProxyBaseURLTPL, addr)
		patch := client.RawPatch(types.MergePatchType, []byte(data))
		if err := svc.kubeClient.Status().Patch(ctx, cr, patch); err != nil {
			log.Error("patch Status.ProxyBaseUrl failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}
	if cr.Status.DataflowAgentUrl == "" && !isHibernated {
		addr, err := svc.getDataflowAgentURL(ctx, cr)
		if err != nil {
			return err
		}
		data := fmt.Sprintf(patchAPPBoxStatusDataflowAgentURLTPL, addr)
		patch := client.RawPatch(types.MergePatchType, []byte(data))
		if err := svc.kubeClient.Status().Patch(ctx, cr, patch); err != nil {
			log.Error("patch Status.DatatflowAgentUrl failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}
	annoMap := map[string]string{}
	k8sVersion := cr.Spec.KubernetesConfig.Version
	dedicatedTemplateID := cr.Spec.KubernetesConfig.DedicatedTemplateID
	dpTemplateID := cr.Spec.KubernetesConfig.DPTemplateID

	dedicatedTemplate, tmplExists := svc.tmplconfig.GetDedicatedTemplateWithVersion(dedicatedTemplateID, k8sVersion)
	if !tmplExists {
		return bizerr.New(bizerr.ConfigInvalid, fmt.Sprintf("Dedicated Template NotFound for templateId %s and version %s", dedicatedTemplateID, k8sVersion))
	}
	dpTemplate, tmplExists := svc.tmplconfig.GetDPTemplateWithVersion(dpTemplateID, k8sVersion)
	if !tmplExists {
		return bizerr.New(bizerr.ConfigInvalid, fmt.Sprintf("Data Platform Template NotFound for templateId %s and version %s", dpTemplateID, k8sVersion))
	}

	if existedTag, err := svc.getDataflowAgentTag(ctx, cr); err != nil {
		return err
	} else {
		values := dpTemplate.DataflowAgent.ToHelmValues()
		image := values["image"]
		generatedTag := image.(map[string]interface{})["tag"].(string)
		if existedTag != generatedTag {
			return bizerr.New(bizerr.ConfigInvalid, fmt.Sprintf("dataflow-agent tag not match, existed:%s, generated:%s", existedTag, generatedTag))
		}
	}
	if existedTag, found, err := svc.getTiFlowOperatorTag(ctx, cr); err != nil {
		return err
	} else {
		if !found {
			annoMap[v1alpha1.AnnotationKeyDisableTiflowOperator] = "true"
		} else {
			values := dpTemplate.TiFlowOperator.ToHelmValues()
			image := values["image"]
			generatedTag := image.(map[string]interface{})["tag"].(string)
			if existedTag != generatedTag {
				return bizerr.New(bizerr.ConfigInvalid, fmt.Sprintf("tiflow-operator tag not match, existed:%s, generated:%s", existedTag, generatedTag))
			}
		}
	}
	if existedTag, err := svc.getClusterProxyTag(ctx, cr); err != nil {
		return err
	} else {
		values := dedicatedTemplate.ClusterProxy.ToHelmValues()
		image := values["image"]
		generatedTag := image.(map[string]interface{})["tag"].(string)
		if existedTag != generatedTag {
			return bizerr.New(bizerr.ConfigInvalid, fmt.Sprintf("cluster-proxy tag not match, existed:%s, generated:%s", existedTag, generatedTag))
		}
	}
	if existedTag, err := svc.getSLIReporterTag(ctx, cr); err != nil {
		return err
	} else {
		values := dedicatedTemplate.SLIReporter.ToHelmValues()
		image := values["image"]
		generatedTag := image.(map[string]interface{})["tag"].(string)
		if existedTag != generatedTag {
			return bizerr.New(bizerr.ConfigInvalid, fmt.Sprintf("sli-reporter tag not match, existed:%s, generated:%s", existedTag, generatedTag))
		}
	}
	switch svc.provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		// overwrite addon version
		annoMap[v1alpha1.AnnotationKeyLegacyAPPBoxSkipAddonAWSEBSCSIDriver] = "true"
		annoMap[v1alpha1.AnnotationKeyLegacyAPPBoxSkipAddonCoreDNS] = "true"
		annoMap[v1alpha1.AnnotationKeyLegacyAPPBoxSkipAddonKubeProxy] = "true"
		annoMap[v1alpha1.AnnotationKeyLegacyAPPBoxSkipAddonVPCCNI] = "true"
	case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
	}
	bs, err := json.Marshal(annoMap)
	if err != nil {
		log.Error("json marshal failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
		return bizerr.New(bizerr.InternalServerError, err.Error())
	}
	data := fmt.Sprintf(patchAnnotationMapTPL, string(bs))
	patch := client.RawPatch(types.MergePatchType, []byte(data))
	if err := svc.kubeClient.Patch(ctx, cr, patch); err != nil {
		log.Error("patch annotation failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}

	return nil
}

func (svc *Service) getObjectInAPPBox(ctx context.Context, appbox *v1alpha1.APPBox, obj client.Object) error {
	key := client.ObjectKeyFromObject(obj).String()
	if err := svc.manager.Register(ctx, appbox); err != nil {
		slog.Error("register appbox to manager falied", "key", key, "error", err)
		return bizerr.New(bizerr.InternalServerError, err.Error())
	}
	getter, err := svc.manager.APIReaderGetter(ctx, client.ObjectKeyFromObject(appbox))
	if err != nil {
		slog.Error("get ClientGetter from manager falied", "key", key, "error", err)
		return bizerr.New(bizerr.InternalServerError, err.Error())
	}
	cli, err := getter.GetAPIReader(ctx)
	if err != nil {
		slog.Error("get Client from getter falied", "key", key, "error", err)
		return bizerr.New(bizerr.InternalServerError, err.Error())
	}
	if err := cli.Get(ctx, client.ObjectKeyFromObject(obj), obj); err != nil {
		slog.Error("get object falied", "key", key, "error", err)
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	return nil
}

func (svc *Service) getObjectInAPPBoxMayNotFound(ctx context.Context, appbox *v1alpha1.APPBox, obj client.Object) (bool, error) {
	key := client.ObjectKeyFromObject(obj).String()
	if err := svc.manager.Register(ctx, appbox); err != nil {
		slog.Error("register appbox to manager falied", "key", key, "error", err)
		return false, bizerr.New(bizerr.InternalServerError, err.Error())
	}
	getter, err := svc.manager.APIReaderGetter(ctx, client.ObjectKeyFromObject(appbox))
	if err != nil {
		slog.Error("get ClientGetter from manager falied", "key", key, "error", err)
		return false, bizerr.New(bizerr.InternalServerError, err.Error())
	}
	cli, err := getter.GetAPIReader(ctx)
	if err != nil {
		slog.Error("get Client from getter falied", "key", key, "error", err)
		return false, bizerr.New(bizerr.InternalServerError, err.Error())
	}
	if err := cli.Get(ctx, client.ObjectKeyFromObject(obj), obj); err != nil {
		if apierrors.IsNotFound(err) {
			return false, nil
		}
		slog.Error("get object falied", "key", key, "error", err)
		return false, bizerr.New(bizerr.K8SClientError, err.Error())
	}
	return true, nil
}

func (svc *Service) getProxyBaseProxyURL(ctx context.Context, appbox *v1alpha1.APPBox) (string, error) {
	dedicatedTemplate, ok := svc.tmplconfig.GetDedicatedTemplateWithVersion(appbox.Spec.KubernetesConfig.DedicatedTemplateID, appbox.Spec.KubernetesConfig.Version)
	if !ok {
		return "", bizerr.New(bizerr.InternalServerError, fmt.Sprintf("dedicatedTempalte for k8s version %v not found", appbox.Spec.KubernetesConfig.Version))
	}
	clusterproxyCfg := dedicatedTemplate.ClusterProxy
	service := new(corev1.Service)
	service.Name = clusterproxyCfg.GetName()
	service.Namespace = clusterproxyCfg.GetNamespace()
	if err := svc.getObjectInAPPBox(ctx, appbox, service); err != nil {
		return "", err
	}
	var addr string
	if service.Spec.Type == corev1.ServiceTypeLoadBalancer {
		if len(service.Status.LoadBalancer.Ingress) < 1 {
			return "", bizerr.New(bizerr.InternalServerError, "Status.LoadBalancer.Ingress for clusterproxy service is empty")
		}
		if svc.provider == consts.ProviderTypeAWS || svc.provider == consts.ProviderTypeEKS {
			addr = service.Status.LoadBalancer.Ingress[0].Hostname
		} else if svc.provider == consts.ProviderTypeGCP || svc.provider == consts.ProviderTypeGKE {
			addr = service.Status.LoadBalancer.Ingress[0].IP
		}
	}
	if addr == "" {
		return "", bizerr.New(bizerr.InternalServerError, "proxyBaseUrl cannot be found")
	}
	return addr, nil
}

func (svc *Service) getDataflowAgentURL(ctx context.Context, appbox *v1alpha1.APPBox) (string, error) {
	dpTemplate, ok := svc.tmplconfig.GetDPTemplateWithVersion(appbox.Spec.KubernetesConfig.DPTemplateID, appbox.Spec.KubernetesConfig.Version)
	if !ok {
		return "", bizerr.New(bizerr.InternalServerError, fmt.Sprintf("dpTempalte for k8s version %v not found", appbox.Spec.KubernetesConfig.Version))
	}
	dataflowagentCfg := dpTemplate.DataflowAgent
	service := new(corev1.Service)
	service.Name = dataflowagentCfg.GetName()
	service.Namespace = dataflowagentCfg.GetNamespace()
	if err := svc.getObjectInAPPBox(ctx, appbox, service); err != nil {
		return "", err
	}
	var addr string
	if service.Spec.Type == corev1.ServiceTypeLoadBalancer {
		if len(service.Status.LoadBalancer.Ingress) < 1 {
			return "", bizerr.New(bizerr.InternalServerError, "Status.LoadBalancer.Ingress for clusterproxy service is empty")
		}
		if svc.provider == consts.ProviderTypeAWS || svc.provider == consts.ProviderTypeEKS {
			addr = service.Status.LoadBalancer.Ingress[0].Hostname
		} else if svc.provider == consts.ProviderTypeGCP || svc.provider == consts.ProviderTypeGKE {
			addr = service.Status.LoadBalancer.Ingress[0].IP
		}
	}
	if addr == "" {
		return "", bizerr.New(bizerr.InternalServerError, "dataflowAgentUrl cannot be found")
	}
	return addr, nil
}

//nolint:unused
func (svc *Service) getCoreDNSAddonVersion(ctx context.Context, appbox *v1alpha1.APPBox) (string, error) {
	deploy := new(appsv1.Deployment)
	deploy.Name = "coredns"
	deploy.Namespace = "kube-system"
	if err := svc.getObjectInAPPBox(ctx, appbox, deploy); err != nil {
		return "", err
	}
	image := deploy.Spec.Template.Spec.Containers[0].Image
	_, addonVersion, ok := strings.Cut(image, ":")
	if !ok {
		slog.Error("infer coredns addon version falied", "appbox", client.ObjectKeyFromObject(appbox).String(), "image", image)
		return "", bizerr.New(bizerr.K8SClientError, "can not found coredns addonversion form image")
	}
	return addonVersion, nil
}

//nolint:unused
func (svc *Service) getKubeProxyAddonVersion(ctx context.Context, appbox *v1alpha1.APPBox) (string, error) {
	ds := new(appsv1.DaemonSet)
	ds.Name = "kube-proxy"
	ds.Namespace = "kube-system"
	if err := svc.getObjectInAPPBox(ctx, appbox, ds); err != nil {
		return "", err
	}
	image := ds.Spec.Template.Spec.Containers[0].Image
	_, addonVersion, ok := strings.Cut(image, ":")
	if !ok {
		slog.Error("infer coredns addon version falied", "appbox", client.ObjectKeyFromObject(appbox).String(), "image", image)
		return "", bizerr.New(bizerr.K8SClientError, "can not found coredns addonversion form image")
	}
	return addonVersion, nil
}

//nolint:unused
func (svc *Service) getVPCCNIAddonVersion(ctx context.Context, appbox *v1alpha1.APPBox) (string, error) {
	ds := new(appsv1.DaemonSet)
	ds.Name = "aws-node"
	ds.Namespace = "kube-system"
	if err := svc.getObjectInAPPBox(ctx, appbox, ds); err != nil {
		return "", err
	}
	image := ds.Spec.Template.Spec.Containers[0].Image
	_, addonVersion, ok := strings.Cut(image, ":")
	if !ok {
		slog.Error("infer coredns addon version falied", "appbox", client.ObjectKeyFromObject(appbox).String(), "image", image)
		return "", bizerr.New(bizerr.K8SClientError, "can not found coredns addonversion form image")
	}
	return addonVersion, nil
}

// in case we need later, fix lint
// func (svc *Service) getAWSEBSCSIDriverAddonInfo(ctx context.Context, appbox *v1alpha1.APPBox) (string, map[string]string, error) {
// 	ds := new(appsv1.DaemonSet)
// 	ds.Name = "ebs-csi-node"
// 	ds.Namespace = "kube-system"
// 	if err := svc.getObjectInAPPBox(ctx, appbox, ds); err != nil {
// 		return "", nil, err
// 	}
// 	image := ds.Spec.Template.Spec.Containers[0].Image
// 	_, addonVersion, ok := strings.Cut(image, ":")
// 	if !ok {
// 		slog.Error("infer coredns addon version falied", "appbox", client.ObjectKeyFromObject(appbox).String(), "image", image)
// 		return "", nil, bizerr.New(bizerr.K8SClientError, "can not found coredns addonversion form image")
// 	}
// 	return addonVersion, ds.Spec.Template.Spec.NodeSelector, nil
// }

func (svc *Service) getDataflowAgentTag(ctx context.Context, appbox *v1alpha1.APPBox) (string, error) {
	deploy := new(appsv1.Deployment)
	deploy.Name = "dataflow-agent"
	deploy.Namespace = "tidb-admin"
	if err := svc.getObjectInAPPBox(ctx, appbox, deploy); err != nil {
		return "", err
	}
	image := deploy.Spec.Template.Spec.Containers[0].Image
	_, tag, ok := strings.Cut(image, ":")
	if !ok {
		slog.Error("infer dataflow-agent tag falied", "appbox", client.ObjectKeyFromObject(appbox).String(), "image", image)
		return "", bizerr.New(bizerr.K8SClientError, "can not found dataflow-agent tag form image")
	}
	return tag, nil
}

func (svc *Service) getTiFlowOperatorTag(ctx context.Context, appbox *v1alpha1.APPBox) (string, bool, error) {
	deploy := new(appsv1.Deployment)
	deploy.Name = "tiflow-operator"
	deploy.Namespace = "tidb-admin"
	found, err := svc.getObjectInAPPBoxMayNotFound(ctx, appbox, deploy)
	if err != nil {
		return "", false, err
	} else if !found {
		return "", false, nil
	}
	image := deploy.Spec.Template.Spec.Containers[0].Image
	_, tag, ok := strings.Cut(image, ":")
	if !ok {
		slog.Error("infer tiflow-operator tag falied", "appbox", client.ObjectKeyFromObject(appbox).String(), "image", image)
		return "", false, bizerr.New(bizerr.K8SClientError, "can not found tiflow-operator tag form image")
	}
	return tag, true, nil
}

func (svc *Service) getClusterProxyTag(ctx context.Context, appbox *v1alpha1.APPBox) (string, error) {
	deploy := new(appsv1.Deployment)
	deploy.Name = "proxy-cluster-proxy"
	deploy.Namespace = "tidb-admin"
	if err := svc.getObjectInAPPBox(ctx, appbox, deploy); err != nil {
		return "", err
	}
	image := deploy.Spec.Template.Spec.Containers[0].Image
	_, tag, ok := strings.Cut(image, ":")
	if !ok {
		slog.Error("infer cluster-proxy tag falied", "appbox", client.ObjectKeyFromObject(appbox).String(), "image", image)
		return "", bizerr.New(bizerr.K8SClientError, "can not found cluster-proxy tag form image")
	}
	return tag, nil
}

func (svc *Service) getSLIReporterTag(ctx context.Context, appbox *v1alpha1.APPBox) (string, error) {
	deploy := new(appsv1.Deployment)
	deploy.Name = "reporter-sli-reporter"
	deploy.Namespace = "tidb-admin"
	if err := svc.getObjectInAPPBox(ctx, appbox, deploy); err != nil {
		return "", err
	}
	image := deploy.Spec.Template.Spec.Containers[0].Image
	_, tag, ok := strings.Cut(image, ":")
	if !ok {
		slog.Error("infer sli-reporter tag falied", "appbox", client.ObjectKeyFromObject(appbox).String(), "image", image)
		return "", bizerr.New(bizerr.K8SClientError, "can not found sli-reporter tag form image")
	}
	return tag, nil
}

func (svc *Service) ImportAPPRoute(ctx context.Context, cr *v1alpha1.APPRoute) error {
	if !cr.DeletionTimestamp.IsZero() {
		return bizerr.New(bizerr.ParameterInvalid, "deletiontimestamp has value")
	}
	// check labels
	var tenantID, projectID, clusterID string
	if clusterID = cr.Labels[v1alpha1.ClusterIDKey]; clusterID != cr.Name {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is mismatched", v1alpha1.ClusterIDKey))
	}
	if cr.Labels[v1alpha1.TagClusterKey] != clusterID {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is mismatched", v1alpha1.TagClusterKey))
	}

	for key, value := range cr.Labels {
		if strings.HasPrefix(key, v1alpha1.TenantIDKeyPrefix) && value == "true" {
			tenantID = strings.TrimPrefix(key, v1alpha1.TenantIDKeyPrefix)
		}
		if strings.HasPrefix(key, v1alpha1.ProjectIDKeyPrefix) && value == "true" {
			projectID = strings.TrimPrefix(key, v1alpha1.ProjectIDKeyPrefix)
		}
	}
	if tenantID == "" {
		return bizerr.New(bizerr.ParameterInvalid, "tenantId label is empty")
	}
	if projectID == "" {
		return bizerr.New(bizerr.ParameterInvalid, "projectId label is empty")
	}
	if cr.Labels[v1alpha1.VPCNameKey] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is mismatched", v1alpha1.VPCNameKey))
	}
	if cr.Spec.VPCRef.Name == "" {
		return bizerr.New(bizerr.ParameterInvalid, "Spec.VPCRef.Name is empty")
	}
	if cr.Spec.APPBoxRef.Name == "" {
		return bizerr.New(bizerr.ParameterInvalid, "Spec.APPBoxRef.Name is empty")
	}

	// set field which caller cannot set.
	cr.Labels[v1alpha1.RegionKey] = svc.region
	cr.Labels = util.MergeMaps(cr.Labels, svc.config.ResourceLabels)
	if cr.Annotations == nil {
		cr.Annotations = map[string]string{}
	}
	cr.Annotations[v1alpha1.AnnotationKeyReconciliationPaused] = "true"
	cr.SetFinalizers([]string{v1alpha1.Finalizer})
	namer := utils.NewNamer(tenantID, projectID, clusterID, svc.config.RegionConfig.Name, svc.config.Env)
	cr.Namespace = namer.GetNamespace()
	cr.Spec.APPBoxRef.Namespace = namer.GetNamespace()
	cr.Spec.VPCRef.Namespace = namer.GetNamespace()
	cr.Status.Phase = v1alpha1.APPRoutePhaseActive
	cr.Status.Reason = "migrated from infra-apiserver"

	if err := upsertObject[*v1alpha1.APPRoute](ctx, svc.kubeClient, svc.kubeAPIReader, cr); err != nil {
		log.Error("update approute failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
		return err
	}
	return nil
}

func (svc *Service) CheckAPPRoute(ctx context.Context, key client.ObjectKey) error {
	cr := new(v1alpha1.APPRoute)
	namer := utils.NewNamer("", "", "", svc.config.RegionConfig.Name, svc.config.Env)
	key.Namespace = namer.GetNamespace()
	if err := svc.kubeAPIReader.Get(ctx, key, cr); err != nil {
		if apierrors.IsNotFound(err) {
			return bizerr.New(bizerr.ResourceNotFound, err.Error())
		}
		log.Error("get approute failed", zap.String("key", key.String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	if ref := cr.Spec.VPCRef; ref != nil && ref.Name != "" {
		vpc := new(v1alpha1.VPC)
		vpc.Namespace = cr.Namespace
		vpc.Name = ref.Name
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(vpc), vpc); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			log.Error("get approute failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}
	if ref := cr.Spec.APPBoxRef; ref != nil && ref.Name != "" {
		appbox := new(v1alpha1.APPBox)
		appbox.Namespace = cr.Namespace
		appbox.Name = ref.Name
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(appbox), appbox); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			log.Error("get appbox failed", zap.String("key", client.ObjectKeyFromObject(appbox).String()), zap.Error(err))
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}
	clusterID := cr.Labels[v1alpha1.ClusterIDKey]
	// o11yroute for tidbcluster use clusterID as cr name
	{
		o11yroute := new(v1alpha1.O11yRoute)
		o11yroute.Name = clusterID
		o11yroute.Namespace = cr.Namespace
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(o11yroute), o11yroute); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			log.Error("get o11yroute failed", zap.String("key", client.ObjectKeyFromObject(o11yroute).String()), zap.Error(err))
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}
	// call cluster-service to check cluster is existed and set ForCluster
	input := &clusterctlpb.GetClusterRequest{
		ClusterId: clusterID,
	}
	cluster, err := svc.clusterClient.GetCluster(ctx, input)
	if err != nil {
		log.Error("call clusterClient.GetCluster failed", zap.Any("input", input), zap.Error(err))
		return bizerr.New(bizerr.InternalServerError, err.Error())
	}
	// some fields are retrieve form other get api
	{
		input := &clusterctlpb.GetVersionAndConfigurationsRequest{
			ClusterId: clusterID,
		}
		out, err := svc.clusterClient.GetVersionAndConfigurations(ctx, input)
		if err != nil {
			log.Error("call clusterClient.GetVersionAndConfigurations failed", zap.Any("input", input), zap.Error(err))
			return bizerr.New(bizerr.InternalServerError, err.Error())
		}
		cluster.ComponentConfigurations = out.ComponentConfigurations
		cluster.ClusterVersion = out.ClusterVersion
	}
	{
		input := &clusterctlpb.ListTidbGroupsRequest{
			ClusterId: clusterID,
		}
		out, err := svc.clusterClient.ListTidbGroups(ctx, input)
		if err != nil {
			log.Error("call clusterClient.ListTidbGroups failed", zap.Any("input", input), zap.Error(err))
			return bizerr.New(bizerr.InternalServerError, err.Error())
		}
		for _, tidbgroup := range out.TidbGroups {
			if tidbgroup.IsDefault {
				cluster.ClusterIdentifierHash = tidbgroup.ClusterIdentifierHash
				break
			}
		}
	}
	{
		input := &clusterctlpb.GetClusterRoleRequest{
			ClusterId: clusterID,
		}
		if _, err := svc.clusterClient.GetClusterRole(ctx, input); err != nil {
			log.Error("call clusterClient.GetClusterRole failed", zap.Any("input", input), zap.Error(err))
			return bizerr.New(bizerr.InternalServerError, err.Error())
		}
	}
	rpbCluster, err := svc.ccpbClusterTorpbCluster(cluster)
	if err != nil {
		log.Error("convert ccpbCluster to rpbCluster failed", zap.Any("ccpbCluster", cluster), zap.Error(err))
		return bizerr.New(bizerr.InternalServerError, err.Error())
	}
	bs, err := svc.marshalOpt.Marshal(rpbCluster)
	if err != nil {
		log.Error("marshal cluster failed", zap.Error(err))
		return bizerr.New(bizerr.InternalServerError, err.Error())
	}
	cr.Spec.ForCluster = string(bs)
	if err := svc.kubeClient.Update(ctx, cr); err != nil {
		log.Error("update approute failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	return nil
}

func (svc *Service) ImportInternet(ctx context.Context, cr *v1alpha1.Internet) error {
	if !cr.DeletionTimestamp.IsZero() {
		return bizerr.New(bizerr.ParameterInvalid, "deletiontimestamp has value")
	}
	// check label and annotation
	var tenantID, projectID string
	for key, value := range cr.Labels {
		if strings.HasPrefix(key, v1alpha1.TenantIDKeyPrefix) && value == "true" {
			tenantID = strings.TrimPrefix(key, v1alpha1.TenantIDKeyPrefix)
		}
		if strings.HasPrefix(key, v1alpha1.ProjectIDKeyPrefix) && value == "true" {
			projectID = strings.TrimPrefix(key, v1alpha1.ProjectIDKeyPrefix)
		}
	}
	if tenantID == "" {
		return bizerr.New(bizerr.ParameterInvalid, "tenantId label is empty")
	}
	if projectID == "" {
		return bizerr.New(bizerr.ParameterInvalid, "projectId label is empty")
	}
	if cr.Labels[v1alpha1.TagClusterKey] != "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is not empty", v1alpha1.TagClusterKey))
	}
	if cr.Labels[v1alpha1.VPCNameKey] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is empty", v1alpha1.VPCNameKey))
	}
	if cr.Spec.VPCRef.Name == "" {
		return bizerr.New(bizerr.ParameterInvalid, "Spec.VPCRef.Name is empty")
	}

	switch svc.provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameVPC] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-vpc anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameInternetGateway] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-internetgateway anno is empty")
		}
		if cr.Spec.Provider != "eks" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Provider is mismatched")
		}
		// the Status.S3GatewayEndpoint.Name must match VPC name for migreated Internet.
		if name := cr.Status.S3GatewayEndpoint.Name; name != "" {
			expectedName := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameVPC] + "-gateway-endpoint-s3"
			if overwriteName := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameS3Gateway]; overwriteName != "" {
				expectedName = overwriteName
			}
			if name != expectedName {
				return bizerr.New(bizerr.ParameterInvalid, "Status.S3GatewayEndpoint.Name is mismatched")
			}
		}
	case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameNetwork] == "" {
			return bizerr.New(bizerr.K8SClientError, "own-crossplane-name-network anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameRouter] == "" {
			return bizerr.New(bizerr.K8SClientError, "own-crossplane-name-router anno is empty")
		}
		if cr.Spec.Provider != "gke" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Provider is mismatched")
		}
	}

	// set fields
	cr.Labels[v1alpha1.IsLegacyKey] = "true"
	cr.Labels[v1alpha1.RegionKey] = svc.region
	cr.Labels = util.MergeMaps(cr.Labels, svc.config.ResourceLabels)
	cr.Annotations[v1alpha1.AnnotationKeyReconciliationPaused] = "true"
	cr.SetFinalizers([]string{v1alpha1.Finalizer})
	namer := utils.NewNamer(tenantID, projectID, "", svc.config.RegionConfig.Name, svc.config.Env)
	cr.Namespace = namer.GetNamespace()
	cr.Spec.VPCRef.Namespace = namer.GetNamespace()
	cr.Spec.Region = svc.region
	cr.Status.Phase = v1alpha1.InternetPhaseActive
	cr.Status.Reason = "migrated from infra-apiserver"

	if err := upsertObject[*v1alpha1.Internet](ctx, svc.kubeClient, svc.kubeAPIReader, cr); err != nil {
		log.Error("update internet failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
		return err
	}
	return nil
}

func (svc *Service) CheckInternet(ctx context.Context, key client.ObjectKey) error {
	cr := new(v1alpha1.Internet)
	namer := utils.NewNamer("", "", "", svc.config.RegionConfig.Name, svc.config.Env)
	key.Namespace = namer.GetNamespace()
	if err := svc.kubeAPIReader.Get(ctx, key, cr); err != nil {
		if apierrors.IsNotFound(err) {
			return bizerr.New(bizerr.ResourceNotFound, err.Error())
		}
		log.Error("get internet failed", zap.String("key", key.String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	// TODO: check crossplane related cr is existed.
	vpc := new(v1alpha1.VPC)
	if ref := cr.Spec.VPCRef; ref != nil && ref.Name != "" {
		vpc.Namespace = ref.Namespace
		vpc.Name = ref.Name
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(vpc), vpc); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}

	switch svc.provider {
	case "aws", "eks":
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameVPC]; name != "" {
			vpc := new(ec2v1beta1.VPC)
			vpc.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(vpc), vpc); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if cr.Status.VPCID != xpmeta.GetExternalName(vpc) {
				return bizerr.New(bizerr.ParameterInvalid, "Status.VPCID is mismatched")
			}
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameInternetGateway]; name != "" {
			igw := new(ec2v1beta1.InternetGateway)
			igw.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(igw), igw); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if cr.Status.InternetGateway.Name != igw.Name {
				return bizerr.New(bizerr.ParameterInvalid, "Status.InternetGateway.Name is mismatched")
			}
			if cr.Status.InternetGateway.InternetGatewayID != igw.Status.AtProvider.InternetGatewayID {
				return bizerr.New(bizerr.ParameterInvalid, "Status.InternetGateway.InternetGatewayID is mismatched")
			}
		}
		// make sure address name is correct
		if len(cr.Status.IPAddresses) > 0 {
			name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameVPC]
			for _, zone := range svc.getZones(vpc) {
				ip := v1alpha1.FindIPAddressByName(cr.Status.IPAddresses, infranamer.PublicSubnetName(name, zone))
				if ip == nil {
					return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("cannot found ip from Status.IPAddresses for zone %v", zone))
				}
			}
		}
		for _, item := range cr.Status.IPAddresses {
			address := new(ec2v1beta1.Address)
			address.Name = item.Name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(address), address); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if item.AllocationID != xpmeta.GetExternalName(address) {
				return bizerr.New(bizerr.ParameterInvalid, "Status.IPAddress.AllocationID is mismatched")
			}
		}
		for _, item := range cr.Status.NATs {
			natgw := new(ec2v1beta1.NATGateway)
			natgw.Name = item.Name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(natgw), natgw); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if item.NATGatewayID != xpmeta.GetExternalName(natgw) {
				return bizerr.New(bizerr.ParameterInvalid, "Status.NATs.NATGatewayID is mismatched")
			}
		}
		for _, item := range cr.Status.RouteTables {
			rt := new(ec2v1beta1.RouteTable)
			rt.Name = item.Name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(rt), rt); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if item.RouteTableID != xpmeta.GetExternalName(rt) {
				return bizerr.New(bizerr.ParameterInvalid, "Status.RouteTables.RouteTableID is mismatched")
			}
		}
		for _, item := range cr.Status.Routes {
			r := new(ec2v1alpha1.Route)
			r.Name = item.Name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(r), r); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if item.RouteTableRef != r.Spec.ForProvider.CustomRouteParameters.RouteTableIDRef.Name {
				return bizerr.New(bizerr.ParameterInvalid, "Status.Routes.RouteTableRef is mismatched")
			}
		}
		if cr.Status.S3GatewayEndpoint.Name != "" {
			vpce := new(ec2v1alpha1.VPCEndpoint)
			vpce.Name = cr.Status.S3GatewayEndpoint.Name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(vpce), vpce); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if cr.Status.S3GatewayEndpoint.EndpointID != meta.GetExternalName(vpce) {
				return bizerr.New(bizerr.ParameterInvalid, "Status.S3GatewayEndpoint.EndpointID is mismatched")
			}
		}
	case "gcp", "gke":
		for _, item := range cr.Status.CloudRouters {
			rr := new(computev1alpha1.Router)
			rr.Name = item.Name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(rr), rr); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if item.CloudRouterID != rr.Status.AtProvider.ID {
				return bizerr.New(bizerr.ParameterInvalid, "Status.CloudRouters.CloudRouterID is mismatched")
			}
			if len(item.NATs) != len(rr.Spec.ForProvider.Nats) {
				return bizerr.New(bizerr.ParameterInvalid, "Status.CloudRouters.NATs is mismatched")
			}
			for i, value := range rr.Spec.ForProvider.Nats {
				nat := item.NATs[i]
				if strings.Join(nat.IPs, ",") != strings.Join(value.NatIps, ",") {
					return bizerr.New(bizerr.ParameterInvalid, "Status.CloudRouters.NATs.IPs is mismatched")
				}
			}
		}
	}
	return nil
}

func (svc *Service) getZones(vpc *v1alpha1.VPC) []string {
	if len(vpc.Spec.Zones) > 0 {
		return vpc.Spec.Zones
	}
	return svc.config.RegionConfig.Zones
}

func (svc *Service) ImportPrivateLinkService(ctx context.Context, cr *v1alpha1.PrivateLinkService) error {
	// v1alpha1.VPCIDKey:          approute.GetLabels()[v1alpha1.VPCIDKey],
	// v1alpha1.VPCNameKey:        approute.GetLabels()[v1alpha1.VPCNameKey],
	// v1alpha1.ClusterIDKey:      req.PrivateLinkService.ClusterId,
	// v1alpha1.LoadBalancerIDKey: req.PrivateLinkService.TidbGroupId,
	if !cr.DeletionTimestamp.IsZero() {
		return bizerr.New(bizerr.ParameterInvalid, "deletiontimestamp has value")
	}
	// check label and annotation
	if cr.Labels[v1alpha1.VPCNameKey] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is empty", v1alpha1.VPCNameKey))
	}
	if cr.Labels[v1alpha1.VPCIDKey] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is empty", v1alpha1.VPCIDKey))
	}
	if cr.Labels[v1alpha1.ClusterIDKey] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is empty", v1alpha1.ClusterIDKey))
	}
	if cr.Labels[v1alpha1.LoadBalancerIDKey] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is empty", v1alpha1.LoadBalancerIDKey))
	}
	if cr.Spec.VPCRef.Name == "" {
		return bizerr.New(bizerr.ParameterInvalid, "Spec.VPCRef.Name is empty")
	}
	// TODO: how to check Spec.LoadBalancer and Spec.Connection

	switch svc.provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameVPCEService]; name == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-vpcesvc anno is empty")
		} else if cr.Status.EndpointService.Name != name {
			return bizerr.New(bizerr.ParameterInvalid, "Status.EndpointService.Name is mismatched")
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameResourceRecordSet]; name == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-rrset anno is empty")
		}
		if cr.Spec.Provider != "eks" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Provider is mismatched")
		}
	case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNamePSCNATSubenet] == "" {
			return bizerr.New(bizerr.K8SClientError, "own-crossplane-name-pscnatsubent anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNamePSCServiceAttachment] == "" {
			return bizerr.New(bizerr.K8SClientError, "own-crossplane-name-pscserviceattachment anno is empty")
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNamePSCDomain]; name == "" {
			return bizerr.New(bizerr.K8SClientError, "own-crossplane-name-pscdomain anno is empty")
		} else if name != cr.Status.ServiceAttachment.ConnectDomain {
			return bizerr.New(bizerr.ParameterInvalid, "Status.ServiceAttachment.ConnectDomain is mismatched")
		}
		if cr.Spec.Provider != "gke" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Provider is mismatched")
		}
		if cr.Status.ServiceAttachment.Port == 0 {
			return bizerr.New(bizerr.ParameterInvalid, "Status.ServiceAttachment.Port is empty")
		}
	}
	// set fields
	cr.Labels[v1alpha1.IsLegacyKey] = "true"
	cr.Labels = util.MergeMaps(cr.Labels, svc.config.ResourceLabels)
	cr.Annotations[v1alpha1.AnnotationKeyReconciliationPaused] = "true"
	cr.SetFinalizers([]string{v1alpha1.Finalizer})
	namer := utils.NewNamer("", "", "", svc.config.RegionConfig.Name, svc.config.Env)
	cr.Namespace = namer.GetNamespace()
	cr.Spec.VPCRef.Namespace = namer.GetNamespace()
	cr.Spec.Region = svc.region
	cr.Status.Phase = v1alpha1.PrivateLinkServicePhaseActive
	cr.Status.Message = "migrated from infra-apiserver"

	if err := upsertObject[*v1alpha1.PrivateLinkService](ctx, svc.kubeClient, svc.kubeAPIReader, cr); err != nil {
		log.Error("update privatelinkservice failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
		return err
	}
	return nil
}

func (svc *Service) CheckPrivateLinkService(ctx context.Context, key client.ObjectKey) error {
	cr := new(v1alpha1.PrivateLinkService)
	namer := utils.NewNamer("", "", "", svc.config.RegionConfig.Name, svc.config.Env)
	key.Namespace = namer.GetNamespace()
	if err := svc.kubeAPIReader.Get(ctx, key, cr); err != nil {
		if apierrors.IsNotFound(err) {
			return bizerr.New(bizerr.ResourceNotFound, err.Error())
		}
		log.Error("get privatelinkservice failed", zap.String("key", key.String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	accountID := ""
	if ref := cr.Spec.VPCRef; ref.Name != "" {
		vpc := new(v1alpha1.VPC)
		vpc.Namespace = ref.Namespace
		vpc.Name = ref.Name
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(vpc), vpc); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
		accountID = vpc.Spec.AccountRef.AccountID
		cr.Status.VPCID = vpc.Status.VPC.VPCID
	}
	// check the Spec.LoadBalancers is correct
	for _, lb := range cr.Spec.LoadBalancers {
		tidbgroup, err := svc.clusterClient.GetTidbGroup(ctx, &clusterctlpb.GetTidbGroupRequest{
			ClusterId:   lb.ClusterID,
			TidbGroupId: lb.ID,
		})
		if err != nil {
			return bizerr.New(bizerr.InternalServerError, err.Error())
		}
		if tidbgroup == nil {
			return bizerr.New(bizerr.ParameterInvalid, "tidbgroup not found")
		}
		if !tidbgroup.EnablePrivateLb || tidbgroup.PrivateLb == nil {
			return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("private loadbalancer not found, clusterId: %s, tidbgroupId: %s", lb.ClusterID, lb.ID))
		}
		phase := v1alpha1.LoadBalancerPhaseActive
		if tidbgroup.PrivateLb.Host == "" || tidbgroup.PrivateLb.CloudResourceId == "" {
			phase = v1alpha1.LoadBalancerPhaseCreating
		}
		cr.Status.LoadBalancers = append(cr.Status.LoadBalancers, v1alpha1.LoadBalancerStatus{
			ID:              tidbgroup.Id,
			CloudResourceId: tidbgroup.PrivateLb.CloudResourceId,
			Port:            tidbgroup.PrivateLb.Port,
			Host:            tidbgroup.PrivateLb.Host,
			Type:            clusterctlpb.LoadBalancer_PRIVATE.String(),
			Phase:           phase,
		})
		cr.Status.DNSIdentify = tidbgroup.ClusterIdentifierHash
	}

	switch svc.provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		// following field can be set by caller
		// Spec.Connections is set by caller from Cluster.ExtraGroups.SqlEndpoints.PrivateLinkService.AWS.AcceptedEndpoints
		// Status.Connections is set by caller from Cluster.Status.SqlEndpoints.PrivateLinkService.AWS.Connections
		// Status.EndpointService is set by caller from Cluster.Status.SqlEndpoints.PrivateLinkService.AWS
		// Status.DNSRecord is set by caller from vpce cp cr Status.AtProvider.ServiceConfiguration.PrivateDNSNameConfiguration
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameVPCEService]; name != "" {
			vpcesvc := new(ec2v1alpha1.VPCEndpointServiceConfiguration)
			vpcesvc.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(vpcesvc), vpcesvc); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if cr.Status.EndpointService.Name != vpcesvc.Name {
				return bizerr.New(bizerr.ParameterInvalid, "Status.EndpointService.Name is mismatched")
			}
			dnsConfig := vpcesvc.Status.AtProvider.ServiceConfiguration.PrivateDNSNameConfiguration
			if cr.Status.DNSRecord.Name != *dnsConfig.Name {
				return bizerr.New(bizerr.ParameterInvalid, "Status.DNSRecord.Name is mismatched")
			}
			if cr.Status.DNSRecord.Status != *dnsConfig.State {
				return bizerr.New(bizerr.ParameterInvalid, "Status.DNSRecord.Status is mismatched")
			}
			if cr.Status.DNSRecord.Type != *dnsConfig.Type {
				return bizerr.New(bizerr.ParameterInvalid, "Status.DNSRecord.Type is mismatched")
			}
			if cr.Status.DNSRecord.Value != *dnsConfig.Value {
				return bizerr.New(bizerr.ParameterInvalid, "Status.DNSRecord.Value is mismatched")
			}
			if cr.Status.DNSOwnership != cr.Status.EndpointService.ServiceID {
				return bizerr.New(bizerr.ParameterInvalid, "Status.DNSOwnership is mismatched")
			}
		}
		{
			recordSet := new(route53v1alpha1.ResourceRecordSet)
			recordSet.Name = cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameResourceRecordSet]
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(recordSet), recordSet); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
		}
		cr.Status.AllowedPrincipals = []string{"*"}
	case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
		// Spec.Connections is set by caller from Cluster.Spec.SqlEndpoints.PrivateLinkService.GCP.AcceptedEndpoints
		// Status.ServiceAttchment is set by caller from Cluster.Status.SqlEndpoints.privateLinkService.GCP
		// check psc subnet is existed via gcp client and set the CIDRBlock to Status
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNamePSCNATSubenet]; name != "" {
			req := new(gcpcomputepb.GetSubnetworkRequest)
			req.Project = accountID
			req.Region = svc.region
			req.Subnetwork = name
			subnetwork, err := svc.subnetworkClient.Get(ctx, req)
			if err != nil {
				return bizerr.New(bizerr.InternalServerError, err.Error())
			}
			cr.Status.NATSubnet.Name = name
			cr.Status.NATSubnet.IPCidrRange = pointer.StringDeref(subnetwork.IpCidrRange, "")
		}

		// Name:              t.serviceAttachmentName(),
		// ServiceAttachment: fmt.Sprintf(serviceAttachmentTemplate, t.VPC.Spec.AccountRef.AccountID, t.PrivateLinkService.Spec.Region, t.serviceAttachmentName()),
		// ConnectDomain:     t.privateServiceConnectDomain(),
		// Port:              t.getServiceAttachmentPort(),
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNamePSCServiceAttachment]; name != "" {
			req := new(gcpcomputepb.GetServiceAttachmentRequest)
			req.Project = accountID
			req.Region = svc.region
			req.ServiceAttachment = name
			serviceattchment, err := svc.serviceattachmentClient.Get(ctx, req)
			if err != nil {
				return bizerr.New(bizerr.InternalServerError, err.Error())
			}
			cr.Status.ServiceAttachment.Name = name
			selflink := "https://www.googleapis.com/compute/v1/" + cr.Status.ServiceAttachment.ServiceAttachment
			if selflink != *serviceattchment.SelfLink {
				return bizerr.New(bizerr.ParameterInvalid, "Status.ServiceAttachment.ServiceAttachment is mismatched")
			}
			// https://cloud.google.com/vpc/docs/configure-private-service-connect-services#gcloud
			// The DNS name might have the format REGION.p.DOMAIN. For example, if the service producer's public domain is example.com, and their published service is in us-west1, then we recommend that they make their service available using us-west1.p.example.com domain names. Include a trailing dot—for example, us-west1.p.example.com.
			if len(serviceattchment.DomainNames) == 0 {
				return bizerr.New(bizerr.InternalServerError, "serviceattchment on gcp is corrupted")
			}
			if cr.Status.ServiceAttachment.ConnectDomain+"." != serviceattchment.DomainNames[0] {
				return bizerr.New(bizerr.ParameterInvalid, "Status.ServiceAttachment.ConnectDomain is mismatched")
			}
		}
	}

	if err := svc.kubeClient.Status().Update(ctx, cr); err != nil {
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	return nil
}

func (svc *Service) ImportVPCPeering(ctx context.Context, cr *v1alpha1.VPCPeering) error {
	if !cr.DeletionTimestamp.IsZero() {
		return bizerr.New(bizerr.ParameterInvalid, "deletiontimestamp has value")
	}
	// name is set by the caller
	// check tenant and project label
	tenantID := cr.Labels[v1alpha1.TenantIDKey]
	projectID := cr.Labels[v1alpha1.ProjectIDKey]
	if tenantID == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is empty", v1alpha1.TenantIDKey))
	}
	if projectID == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is empty", v1alpha1.ProjectIDKey))
	}
	if cr.Labels[v1alpha1.TagClusterKey] != "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is not empty", v1alpha1.TagClusterKey))
	}
	if cr.Annotations[v1alpha1.AnnotationKeySourceVPCID] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s anno is empty", v1alpha1.AnnotationKeySourceVPCID))
	}
	if cr.Annotations[v1alpha1.AnnotationKeySourceVPCCIDR] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s anno is empty", v1alpha1.AnnotationKeySourceVPCCIDR))
	}
	if cr.Annotations[v1alpha1.AnnotationKeySourceAccountID] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s anno is empty", v1alpha1.AnnotationKeySourceAccountID))
	}
	if cr.Annotations[cloudeksv1alpha1.OwnCrossplaneNameAddKey] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s anno is empty", cloudeksv1alpha1.OwnCrossplaneNameAddKey))
	}
	if cr.Spec.SourceVPCRef.Name == "" {
		return bizerr.New(bizerr.ParameterInvalid, "Spec.SourceVPCRef.Name is empty")
	}
	if cr.Status.ExternalName == "" {
		return bizerr.New(bizerr.ParameterInvalid, "Status.ExternalName is empty")
	}

	switch svc.provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		if cr.Spec.Provider != "eks" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Provider is mismatched")
		}
	case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
		if cr.Spec.Provider != "gke" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Provider is mismatched")
		}
	}
	// set some fields which the caller can not set
	cr.Labels[v1alpha1.IsLegacyKey] = "true"
	cr.Labels = util.MergeMaps(cr.Labels, svc.config.ResourceLabels)
	if cr.Annotations == nil {
		cr.Annotations = map[string]string{}
	}
	cr.Annotations[v1alpha1.AnnotationKeyReconciliationPaused] = "true"
	cr.Annotations[v1alpha1.AnnotationKeyRouteTableEnsured] = "true"
	cr.SetFinalizers([]string{v1alpha1.FinalizerVPCPeering})
	namer := utils.NewNamer("", "", "", svc.config.RegionConfig.Name, svc.config.Env)
	cr.Namespace = namer.GetNamespace()
	cr.Spec.SourceVPCRef.Namespace = namer.GetNamespace()
	cr.Spec.Region = svc.region
	cr.Status.Reason = "migrated from infra-apiserver"

	if err := upsertObject[*v1alpha1.VPCPeering](ctx, svc.kubeClient, svc.kubeAPIReader, cr); err != nil {
		log.Error("update vpcpeering failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
		return err
	}
	return nil
}

func (svc *Service) CheckVPCPeering(ctx context.Context, key client.ObjectKey) error {
	cr := new(v1alpha1.VPCPeering)
	namer := utils.NewNamer("", "", "", svc.config.RegionConfig.Name, svc.config.Env)
	key.Namespace = namer.GetNamespace()
	if err := svc.kubeAPIReader.Get(ctx, key, cr); err != nil {
		if apierrors.IsNotFound(err) {
			return bizerr.New(bizerr.ResourceNotFound, err.Error())
		}
		log.Error("get vpcpeering failed", zap.String("key", key.String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	// TODO: check crossplane related cr is existed.
	if ref := cr.Spec.SourceVPCRef; ref != nil && ref.Name != "" {
		vpc := new(v1alpha1.VPC)
		vpc.Namespace = cr.Namespace
		vpc.Name = ref.Name
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(vpc), vpc); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
		// check if the vpc info match the info in annotation
		if cr.Annotations[v1alpha1.AnnotationKeySourceVPCID] != vpc.Status.VPC.VPCID {
			return bizerr.New(bizerr.K8SClientError, "sourceVpcId anno is mismatched")
		}
		if cr.Annotations[v1alpha1.AnnotationKeySourceVPCCIDR] != vpc.Spec.CIDR {
			return bizerr.New(bizerr.K8SClientError, "sourceVpcCIDR anno is mismatched")
		}
		if cr.Annotations[v1alpha1.AnnotationKeySourceAccountID] != vpc.Spec.AccountRef.AccountID {
			return bizerr.New(bizerr.K8SClientError, "sourceVpcAccountId anno is mismatched")
		}
		cr.Spec.AccountRef = vpc.Spec.AccountRef
	}
	xpname := path.Base(cr.Status.CrossplaneResourceName)
	switch svc.provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		peeringconn := new(ec2v1alpha1.VPCPeeringConnection)
		peeringconn.Name = xpname
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(peeringconn), peeringconn); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
		if xpmeta.GetExternalName(peeringconn) != cr.Status.ExternalName {
			return bizerr.New(bizerr.K8SClientError, "externalname is mismatched")
		}
	case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
		peering := new(vpcpeeringv1beta1.Peering)
		peering.Name = xpname
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(peering), peering); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
		if xpmeta.GetExternalName(peering) != cr.Status.ExternalName {
			return bizerr.New(bizerr.K8SClientError, "externalname is mismatched")
		}
	}
	// we need to update the accountRef
	if err := svc.kubeClient.Update(ctx, cr); err != nil {
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	return nil
}

func (svc *Service) ImportVPC(ctx context.Context, cr *v1alpha1.VPC) error {
	if !cr.DeletionTimestamp.IsZero() {
		return bizerr.New(bizerr.ParameterInvalid, "deletiontimestamp has value")
	}
	// check label and annotation
	var tenantID, projectID string
	for key, value := range cr.Labels {
		if strings.HasPrefix(key, v1alpha1.TenantIDKeyPrefix) && value == "true" {
			tenantID = strings.TrimPrefix(key, v1alpha1.TenantIDKeyPrefix)
		}
		if strings.HasPrefix(key, v1alpha1.ProjectIDKeyPrefix) && value == "true" {
			projectID = strings.TrimPrefix(key, v1alpha1.ProjectIDKeyPrefix)
		}
	}
	if tenantID == "" {
		return bizerr.New(bizerr.ParameterInvalid, "tenantId label is empty")
	}
	if projectID == "" {
		return bizerr.New(bizerr.ParameterInvalid, "projectId label is empty")
	}
	if cr.Labels[v1alpha1.TagClusterKey] != "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is not empty", v1alpha1.TagClusterKey))
	}
	if cr.Labels[v1alpha1.VPCIDKey] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is empty", v1alpha1.VPCIDKey))
	}
	if cr.Spec.AccountRef.ProviderConfigName == "" {
		return bizerr.New(bizerr.ParameterInvalid, "Spec.AccountRef.ProviderConfigName label is empty")
	}
	isHibernated := cr.Annotations[v1alpha1.AnnotationKeyIsHiberated] == "true"
	switch svc.provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameVPC] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-vpc anno is empty")
		}
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameSecurityGroup] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-security anno is empty")
		}
		if cr.Spec.AccountRef.AccountID == "" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.AccountRef.AccountID is empty")
		}
		if cr.Spec.AccountRef.Role == "" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.AccountRef.Role is empty")
		}
		providerConfig, err := svc.checkProviderConfigExistedAWS(ctx, cr.Spec.AccountRef.ProviderConfigName)
		if err != nil {
			return err
		}
		if providerConfig.Spec.AssumeRoleARN == nil {
			return bizerr.New(bizerr.InternalServerError, fmt.Sprintf("providerConfig %s Spec.AssumeRoleARN is empty", providerConfig.Name))
		}
		found := roleArnRE.FindStringSubmatch(*providerConfig.Spec.AssumeRoleARN)
		if len(found) != 3 {
			return bizerr.New(bizerr.InternalServerError, "rolearn is invalid")
		}
		account := found[1]
		role := found[2]
		if cr.Spec.AccountRef.AccountID != account {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.AccountRef.AccountID is mismathed")
		}
		if cr.Spec.AccountRef.Role != role {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.AccountRef.Role is mismatched")
		}
		if cr.Spec.Provider != "eks" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Provider is mismatched")
		}
		cr.Labels[v1alpha1.AccountIDKey] = account
	case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
		if cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameNetwork] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-network anno is empty")
		}
		if !isHibernated && cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNamePSCFirewall] == "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-pscfirewall anno is empty")
		}
		if isHibernated && cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNamePSCFirewall] != "" {
			return bizerr.New(bizerr.ParameterInvalid, "own-crossplane-name-pscfirewall anno is set under hibernated")
		}
		if cr.Spec.AccountRef.AccountID == "" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.AccountRef.AccountID is empty")
		}
		providerConfig, err := svc.checkProviderConfigExistedGCP(ctx, cr.Spec.AccountRef.ProviderConfigName)
		if err != nil {
			return err
		}
		if cr.Spec.AccountRef.AccountID != providerConfig.Spec.ProjectID {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.AccountRef.AccountID is mismatched")
		}
		if cr.Spec.Provider != "gke" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Provider is mismatched")
		}
		cr.Labels[v1alpha1.AccountIDKey] = providerConfig.Spec.ProjectID
	}
	// set fields
	cr.Labels[v1alpha1.IsLegacyKey] = "true"
	cr.Labels[v1alpha1.RegionKey] = svc.region
	cr.Labels = util.MergeMaps(cr.Labels, svc.config.ResourceLabels)
	cr.Annotations[v1alpha1.AnnotationKeyReconciliationPaused] = "true"
	cr.SetFinalizers([]string{v1alpha1.Finalizer})
	namer := utils.NewNamer(tenantID, projectID, "", svc.config.RegionConfig.Name, svc.config.Env)
	cr.Namespace = namer.GetNamespace()
	cr.Spec.Region = svc.region
	cr.Status.Phase = v1alpha1.VPCPhaseActive
	cr.Status.Reason = "migrated from infra-apiserver"

	if err := upsertObject[*v1alpha1.VPC](ctx, svc.kubeClient, svc.kubeAPIReader, cr); err != nil {
		log.Error("upsert appbox failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
		return err
	}

	// after vpc create successfully, create a binding cm if not existed
	bindingConfigMap := new(corev1.ConfigMap)
	bindingConfigMap.Namespace = constants.DataPlaneAccountNamespace
	bindingConfigMap.Name = (models.AccountBindingTarget{
		TargetType: constants.AccountBindingVPC,
		TargetKey:  cr.Name,
	}).UniqueTargetName()
	if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(bindingConfigMap), bindingConfigMap); err != nil {
		if !apierrors.IsNotFound(err) {
			log.Error("get binding cm failed", zap.String("key", client.ObjectKeyFromObject(bindingConfigMap).String()), zap.Error(err))
			return err
		}
		configData := map[string]string{}
		configData[constants.DataPlaneAccountConfigMapDefaultKey] = cr.Spec.AccountRef.AccountID
		bindingConfigMap.Labels = svc.config.ResourceLabels
		bindingConfigMap.Data = configData
		if err := svc.kubeClient.Create(ctx, bindingConfigMap); err != nil {
			log.Error("create binding cm failed", zap.String("key", client.ObjectKeyFromObject(bindingConfigMap).String()), zap.Error(err))
			return err
		}
	}
	return nil
}

func (svc *Service) CheckVPC(ctx context.Context, key client.ObjectKey) error {
	cr := new(v1alpha1.VPC)
	namer := utils.NewNamer("", "", "", svc.config.RegionConfig.Name, svc.config.Env)
	key.Namespace = namer.GetNamespace()
	if err := svc.kubeAPIReader.Get(ctx, key, cr); err != nil {
		if apierrors.IsNotFound(err) {
			return bizerr.New(bizerr.ResourceNotFound, err.Error())
		}
		log.Error("get rs vpc failed", zap.String("key", key.String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	switch svc.provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		if _, err := svc.checkProviderConfigExistedAWS(ctx, cr.Spec.AccountRef.ProviderConfigName); err != nil {
			return err
		}
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameVPC]; name != "" {
			vpc := new(ec2v1beta1.VPC)
			vpc.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(vpc), vpc); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get vpc failed", zap.String("key", client.ObjectKeyFromObject(vpc).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if cr.Labels[v1alpha1.VPCIDKey] != xpmeta.GetExternalName(vpc) {
				return bizerr.New(bizerr.ParameterInvalid, "vpcId label is mismatched")
			}
		}
		// check securitygroup and set status
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameSecurityGroup]; name != "" {
			securitygroup := new(ec2v1beta1.SecurityGroup)
			securitygroup.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(securitygroup), securitygroup); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get securitygroup failed", zap.String("key", client.ObjectKeyFromObject(securitygroup).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			// the Status.SecurityGroup field should set by caller.
			// need to check the value is match here
			if cr.Status.SecurityGroup.Name != securitygroup.Name {
				return bizerr.New(bizerr.ParameterInvalid, "Status.SecurityGroup.Name is mismatched")
			}
			if cr.Status.SecurityGroup.SecurityGroupID != securitygroup.Status.AtProvider.SecurityGroupID {
				return bizerr.New(bizerr.ParameterInvalid, "Status.SecurityGroup.SecurityGroupID is mismatched")
			}
		}
		for _, item := range cr.Status.PrivateSubnets {
			// here is how status is set in subnet task
			// SubnetID:           subnet.Status.AtProvider.SubnetID,
			// Name:               subnet.Name,
			// AvailabilityZone:   *subnet.Spec.ForProvider.AvailabilityZone,
			// AvailabilityZoneID: *subnet.Spec.ForProvider.AvailabilityZoneID,
			// CIDRBlock:          subnet.Spec.ForProvider.CIDRBlock,
			subnet := new(ec2v1beta1.Subnet)
			subnet.Name = item.Name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(subnet), subnet); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get subnet failed", zap.String("key", client.ObjectKeyFromObject(subnet).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if item.SubnetID != subnet.Status.AtProvider.SubnetID {
				return bizerr.New(bizerr.ParameterInvalid, "Status.PrivateSubnets.SubnetID is mismatched")
			}
			if item.AvailabilityZone != *subnet.Spec.ForProvider.AvailabilityZone {
				return bizerr.New(bizerr.ParameterInvalid, "Status.PrivateSubnets.AvailabilityZone is mismatched")
			}
			if item.AvailabilityZoneID != *subnet.Spec.ForProvider.AvailabilityZoneID {
				return bizerr.New(bizerr.ParameterInvalid, "Status.PrivateSubnets.AvailabilityZoneID is mismatched")
			}
			if item.CIDRBlock != subnet.Spec.ForProvider.CIDRBlock {
				return bizerr.New(bizerr.ParameterInvalid, "Status.PrivateSubnets.CIDRBlock is mismatched")
			}
		}
		for _, item := range cr.Status.PublicSubnets {
			subnet := new(ec2v1beta1.Subnet)
			subnet.Name = item.Name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(subnet), subnet); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get subnet failed", zap.String("key", client.ObjectKeyFromObject(subnet).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if item.SubnetID != subnet.Status.AtProvider.SubnetID {
				return bizerr.New(bizerr.ParameterInvalid, "Status.PublicSubnets.SubnetID is mismatched")
			}
			if item.AvailabilityZone != *subnet.Spec.ForProvider.AvailabilityZone {
				return bizerr.New(bizerr.ParameterInvalid, "Status.PublicSubnets.AvailabilityZone is mismatched")
			}
			if item.AvailabilityZoneID != *subnet.Spec.ForProvider.AvailabilityZoneID {
				return bizerr.New(bizerr.ParameterInvalid, "Status.PublicSubnets.AvailabilityZoneID is mismatched")
			}
			if item.CIDRBlock != subnet.Spec.ForProvider.CIDRBlock {
				return bizerr.New(bizerr.ParameterInvalid, "Status.PublicSubnets.CIDRBlock is mismatched")
			}
		}
		for _, item := range cr.Status.NodeSubnets {
			subnet := new(ec2v1beta1.Subnet)
			subnet.Name = item.Name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(subnet), subnet); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get subnet failed", zap.String("key", client.ObjectKeyFromObject(subnet).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if item.SubnetID != subnet.Status.AtProvider.SubnetID {
				return bizerr.New(bizerr.ParameterInvalid, "Status.NodeSubnets.SubnetID is mismatched")
			}
			if item.AvailabilityZone != *subnet.Spec.ForProvider.AvailabilityZone {
				return bizerr.New(bizerr.ParameterInvalid, "Status.NodeSubnets.AvailabilityZone is mismatched")
			}
			if item.AvailabilityZoneID != *subnet.Spec.ForProvider.AvailabilityZoneID {
				return bizerr.New(bizerr.ParameterInvalid, "Status.NodeSubnets.AvailabilityZoneID is mismatched")
			}
			if item.CIDRBlock != subnet.Spec.ForProvider.CIDRBlock {
				return bizerr.New(bizerr.ParameterInvalid, "Status.NodeSubnets.CIDRBlock is mismatched")
			}
		}

	case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
		if _, err := svc.checkProviderConfigExistedGCP(ctx, cr.Spec.AccountRef.ProviderConfigName); err != nil {
			return err
		}
		// check xp network existed if annotation is set
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameNetwork]; name != "" {
			network := new(computev1beta1.Network)
			network.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(network), network); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get network failed", zap.String("key", client.ObjectKeyFromObject(network).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if cr.Labels[v1alpha1.VPCIDKey] != xpmeta.GetExternalName(network) {
				return bizerr.New(bizerr.ParameterInvalid, "vpcId label is mismatched")
			}
		}
		// check xp firewall existed if annotation is set
		if name := cr.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNamePSCFirewall]; name != "" {
			firewall := new(computev1alpha1.Firewall)
			firewall.Name = name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(firewall), firewall); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get firewall failed", zap.String("key", client.ObjectKeyFromObject(firewall).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if cr.Status.PSCFirewall.Name != firewall.Name {
				return bizerr.New(bizerr.ParameterInvalid, "Status.PSCFirewall.Name is mismatched")
			}
			if cr.Status.PSCFirewall.FirewallID != firewall.Status.AtProvider.ID {
				return bizerr.New(bizerr.ParameterInvalid, "Status.PSCFirewall.FirewallID is mismatched")
			}
			if len(cr.Status.PSCFirewall.SourceRanges) != len(firewall.Spec.ForProvider.SourceRanges) {
				return bizerr.New(bizerr.ParameterInvalid, "Status.PSCFirewall.SourceRanges length is mismatched")
			}
			if strings.Join(cr.Status.PSCFirewall.SourceRanges, ",") != strings.Join(firewall.Spec.ForProvider.SourceRanges, ",") {
				return bizerr.New(bizerr.ParameterInvalid, "Status.PSCFirewall.SourceRanges is mismatched")
			}
		}
		// check node subnet cr
		for _, item := range cr.Status.NodeSubnets {
			// here is how status is set in subnet task
			// SubnetID:  subnet.Name,
			// Name:      subnet.Name,
			// CIDRBlock: subnet.Spec.ForProvider.IPCidrRange,
			subnetwork := new(computev1beta1.Subnetwork)
			subnetwork.Name = item.Name
			if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(subnetwork), subnetwork); err != nil {
				if apierrors.IsNotFound(err) {
					return bizerr.New(bizerr.ResourceNotFound, err.Error())
				}
				log.Error("get subnetwork failed", zap.String("key", client.ObjectKeyFromObject(subnetwork).String()), zap.Error(err))
				return bizerr.New(bizerr.K8SClientError, err.Error())
			}
			if item.SubnetID != subnetwork.Name {
				return bizerr.New(bizerr.ParameterInvalid, "Status.NodeSubnets.SubnetID is mismatched")
			}
			if item.CIDRBlock != subnetwork.Spec.ForProvider.IPCidrRange {
				return bizerr.New(bizerr.ParameterInvalid, "Status.NodeSubnets.CIDRBlock is mismatched")
			}
		}
	}
	return nil
}

func (svc *Service) ImportO11yRoute(ctx context.Context, cr *v1alpha1.O11yRoute) error {
	if !cr.DeletionTimestamp.IsZero() {
		return bizerr.New(bizerr.ParameterInvalid, "deletiontimestamp has value")
	}
	isForDataPlane := cr.Spec.ApplicationID != "k8s-infra"
	// check labels
	var tenantID, projectID, clusterID string
	for key, value := range cr.Labels {
		if strings.HasPrefix(key, v1alpha1.TenantIDKeyPrefix) && value == "true" {
			tenantID = strings.TrimPrefix(key, v1alpha1.TenantIDKeyPrefix)
		}
		if strings.HasPrefix(key, v1alpha1.ProjectIDKeyPrefix) && value == "true" {
			projectID = strings.TrimPrefix(key, v1alpha1.ProjectIDKeyPrefix)
		}
	}
	if tenantID == "" {
		return bizerr.New(bizerr.ParameterInvalid, "tenantId label is empty")
	}
	if cr.Labels[v1alpha1.TagTenantKey] != tenantID {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is mismatched", v1alpha1.TagTenantKey))
	}
	if projectID == "" {
		return bizerr.New(bizerr.ParameterInvalid, "projectId label is empty")
	}
	if cr.Labels[v1alpha1.TagProjectKey] != projectID {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is mismatched", v1alpha1.TagProjectKey))
	}
	if cr.Labels[v1alpha1.VPCNameKey] == "" {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is empty", v1alpha1.VPCNameKey))
	}
	if cr.Labels[v1alpha1.VPCNameKey] != cr.Spec.VPCRef.Name {
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is mismatched", v1alpha1.VPCNameKey))
	}
	if cr.Spec.ApplicationID == "" {
		return bizerr.New(bizerr.ParameterInvalid, "Spec.ApplicationID is empty")
	}
	if cr.Spec.VPCRef.Name == "" {
		return bizerr.New(bizerr.ParameterInvalid, "Spec.VPCRef.Name is empty")
	}
	if cr.Spec.APPBoxRef.Name == "" {
		return bizerr.New(bizerr.ParameterInvalid, "Spec.APPBoxRef.Name is empty")
	}
	if isForDataPlane {
		clusterID = cr.Labels[v1alpha1.ClusterIDKey]
		if clusterID == "" {
			return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is empty", v1alpha1.ClusterIDKey))
		}
		if cr.Labels[v1alpha1.TagClusterKey] != clusterID {
			return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("%s label is mismatched", v1alpha1.TagClusterKey))
		}
		if cr.Spec.APPRouteRef.Name == "" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.APPRouteRef.Name is empty")
		}
		if cr.Spec.ApplicationID != clusterID {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.ApplicationID is mismatched")
		}
		if cr.Name != clusterID {
			return bizerr.New(bizerr.ParameterInvalid, "Name is mismatched")
		}
		if agent := cr.Spec.Agent; agent == nil {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Agent is empty")
		}
		if cr.Spec.Agent.TargetClusterVersion == "" {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Agent.TargetClusterVersion is empty")
		}
		if topo := cr.Spec.Agent.TargetClusterTopology; topo == nil {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Agent.TargetClusterTopology is empty")
		} else if topo.PD.Replicas == 0 {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Agent.TargetClusterTopology.PD.Replicas is empty")
		} else if topo.TiDB.Replicas == 0 {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Agent.TargetClusterTopology.TiDB.Replicas is empty")
		} else if topo.TiKV.Replicas == 0 {
			return bizerr.New(bizerr.ParameterInvalid, "Spec.Agent.TargetClusterTopology.TiKV.Replicas is empty")
		}
	}
	// set field which caller cannot set.
	cr.Labels[v1alpha1.RegionKey] = svc.region
	cr.Labels = util.MergeMaps(cr.Labels, svc.config.ResourceLabels)
	if cr.Annotations == nil {
		cr.Annotations = map[string]string{}
	}
	cr.Annotations[v1alpha1.AnnotationKeyReconciliationPaused] = "true"
	cr.SetFinalizers([]string{v1alpha1.Finalizer})
	namer := utils.NewNamer(tenantID, projectID, clusterID, svc.config.RegionConfig.Name, svc.config.Env)
	cr.Namespace = namer.GetNamespace()
	cr.Spec.APPBoxRef.Namespace = namer.GetNamespace()
	cr.Spec.VPCRef.Namespace = namer.GetNamespace()
	if isForDataPlane {
		cr.Spec.APPRouteRef.Namespace = namer.GetNamespace()
	}
	cr.Status.Phase = v1alpha1.O11yRoutePhaseActive
	cr.Status.Reason = "migrated from infra-apiserver"

	if err := upsertObject[*v1alpha1.O11yRoute](ctx, svc.kubeClient, svc.kubeAPIReader, cr); err != nil {
		log.Error("update o11yroute failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
		return err
	}
	return nil
}

func (svc *Service) CheckO11yRoute(ctx context.Context, key client.ObjectKey) error {
	cr := new(v1alpha1.O11yRoute)
	namer := utils.NewNamer("", "", "", svc.config.RegionConfig.Name, svc.config.Env)
	key.Namespace = namer.GetNamespace()
	if err := svc.kubeAPIReader.Get(ctx, key, cr); err != nil {
		if apierrors.IsNotFound(err) {
			return bizerr.New(bizerr.ResourceNotFound, err.Error())
		}
		log.Error("get o11yroute failed", zap.String("key", key.String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	if ref := cr.Spec.VPCRef; ref.Name != "" {
		vpc := new(v1alpha1.VPC)
		vpc.Namespace = ref.Namespace
		vpc.Name = ref.Name
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(vpc), vpc); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			log.Error("get rs vpc failed", zap.String("key", client.ObjectKeyFromObject(vpc).String()), zap.Error(err))
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}
	if ref := cr.Spec.APPBoxRef; ref.Name != "" {
		appbox := new(v1alpha1.APPBox)
		appbox.Namespace = ref.Namespace
		appbox.Name = ref.Name
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(appbox), appbox); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			log.Error("get appbox failed", zap.String("key", client.ObjectKeyFromObject(appbox).String()), zap.Error(err))
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}
	if ref := cr.Spec.APPRouteRef; ref != nil {
		approute := new(v1alpha1.APPRoute)
		approute.Namespace = ref.Namespace
		approute.Name = ref.Name
		if err := svc.kubeAPIReader.Get(ctx, client.ObjectKeyFromObject(approute), approute); err != nil {
			if apierrors.IsNotFound(err) {
				return bizerr.New(bizerr.ResourceNotFound, err.Error())
			}
			log.Error("get approute failed", zap.String("key", client.ObjectKeyFromObject(approute).String()), zap.Error(err))
			return bizerr.New(bizerr.K8SClientError, err.Error())
		}
	}
	return nil
}

func (svc *Service) PauseReconcile(ctx context.Context, cr client.Object, pause bool) error {
	namer := utils.NewNamer("", "", "", svc.config.RegionConfig.Name, svc.config.Env)
	cr.SetNamespace(namer.GetNamespace())
	var patch client.Patch
	data := []byte(fmt.Sprintf(patchAnnotationTPL,
		v1alpha1.AnnotationKeyReconciliationPaused,
		strconv.FormatBool(pause)))
	patch = client.RawPatch(types.MergePatchType, data)
	if err := svc.kubeClient.Patch(ctx, cr, patch); err != nil {
		log.Error("patch pause annotation failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	return nil
}

func (svc *Service) SetAnnoSkipGC(ctx context.Context, cr client.Object, value bool) error {
	namer := utils.NewNamer("", "", "", svc.config.RegionConfig.Name, svc.config.Env)
	cr.SetNamespace(namer.GetNamespace())
	var patch client.Patch
	data := []byte(fmt.Sprintf(patchAnnotationTPL,
		v1alpha1.SkipGCAnnotationKey,
		strconv.FormatBool(value)))
	patch = client.RawPatch(types.MergePatchType, data)
	if err := svc.kubeClient.Patch(ctx, cr, patch); err != nil {
		log.Error("patch skipGC annotation failed", zap.String("key", client.ObjectKeyFromObject(cr).String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	return nil
}

func (svc *Service) CheckEKSClusterSpec(ctx context.Context, clusterID string, spec *cloudv1.EKSClusterSpec) error {
	opt := client.MatchingLabels{}
	opt[v1alpha1.TagClusterKey] = clusterID
	list := new(cloudv1.ClusterList)
	if err := svc.kubeAPIReader.List(ctx, list, opt); err != nil {
		log.Error("list cluster failed", zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	if len(list.Items) == 0 {
		return bizerr.New(bizerr.ResourceNotFound, "cluster not found")
	}
	cluster := &(list.Items[0])
	existed, err := cloudv1.EKSClusterSpecForCluster(cluster)
	if err != nil {
		log.Error("get eks spec failed", zap.String("cluster", client.ObjectKeyFromObject(cluster).String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	if spec.Security != nil && spec.Security.MasterKeyARN != nil {
		if existed.Security == nil {
			return bizerr.New(bizerr.K8SClientError, "forProvider.security is empty on migrated cluster CR")
		}
		if existed.Security.MasterKeyARN == nil {
			return bizerr.New(bizerr.K8SClientError, "forProvider.security.masterKeyArn is empty on migrated cluster CR")
		}
		left := *existed.Security.MasterKeyARN
		right := *spec.Security.MasterKeyARN
		if left != right {
			log.Error("detect diff on Spec.ForPorvider.Security.MasterKeyARN",
				zap.String("key", client.ObjectKeyFromObject(cluster).String()),
				zap.String("left", left),
				zap.String("right", right),
			)
			return bizerr.New(bizerr.K8SClientError, "forProvider.security.masterKeyArn is diff on migrated cluster CR")
		}
	}
	return nil
}

func (svc *Service) CheckGKEClusterSpec(ctx context.Context, clusterID string, spec *cloudv1.GKEClusterSpec) error {
	opt := client.MatchingLabels{}
	opt[v1alpha1.TagClusterKey] = clusterID
	list := new(cloudv1.ClusterList)
	if err := svc.kubeAPIReader.List(ctx, list, opt); err != nil {
		log.Error("list cluster failed", zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	if len(list.Items) == 0 {
		return bizerr.New(bizerr.ResourceNotFound, "cluster not found")
	}
	cluster := &(list.Items[0])
	existed, err := cloudv1.GKEClusterSpecForCluster(cluster)
	if err != nil {
		log.Error("get gke spec failed", zap.String("cluster", client.ObjectKeyFromObject(cluster).String()), zap.Error(err))
		return bizerr.New(bizerr.K8SClientError, err.Error())
	}
	if spec.Security != nil && spec.Security.MasterKeyARN != nil {
		if existed.Security == nil {
			return bizerr.New(bizerr.K8SClientError, "forProvider.security is empty on migrated cluster CR")
		}
		if existed.Security.MasterKeyARN == nil {
			return bizerr.New(bizerr.K8SClientError, "forProvider.security.masterKeyArn is empty on migrated cluster CR")
		}
		left := *existed.Security.MasterKeyARN
		right := *spec.Security.MasterKeyARN
		if left != right {
			log.Error("detect diff on Spec.ForPorvider.Security.MasterKeyARN",
				zap.String("key", client.ObjectKeyFromObject(cluster).String()),
				zap.String("left", left),
				zap.String("right", right),
			)
			return bizerr.New(bizerr.K8SClientError, "forProvider.security.masterKeyArn is diff on migrated cluster CR")
		}
	}
	return nil
}

func (svc *Service) ccpbClusterTorpbCluster(cluster *clusterctlpb.Cluster) (*regionalpb.Cluster, error) {
	rpbCluster := new(regionalpb.Cluster)
	bs, err := protojson.Marshal(cluster)
	if err != nil {
		return nil, err
	}
	if err := svc.unmarshalOpt.Unmarshal(bs, rpbCluster); err != nil {
		return nil, err
	}
	rpbCluster.ClusterPhase = regionalpb.Cluster_SYNCED
	return rpbCluster, nil
}