package cloud

import (
	"context"
	"fmt"
	"slices"
	"sort"
	"strings"

	aws_sdk "github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/autoscaling"
	"github.com/aws/aws-sdk-go/service/ec2"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/elliotchance/pie/v2"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	regionalpb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/scheme"
	"github.com/tidbcloud/infra-provider/pkg/util"
	cdcv1 "github.com/tidbcloud/ticdc-service/api/v1"
	"github.com/tidbcloud/ticdc-service/pkg/config"
	ticdc_consts "github.com/tidbcloud/ticdc-service/pkg/consts"
	"github.com/tidbcloud/ticdc-service/pkg/regionalservice"
	ec2v1beta1 "github.com/upbound/provider-aws/apis/ec2/v1beta1"
	eksv1beta1 "github.com/upbound/provider-aws/apis/eks/v1beta1"
	route53v1beta1 "github.com/upbound/provider-aws/apis/route53/v1beta1"
	awsv1beta1 "github.com/upbound/provider-aws/apis/v1beta1"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	corev1 "k8s.io/api/core/v1"
	storagev1 "k8s.io/api/storage/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	// crOwnerRefLabel is the label for all controller created cross-plane resources
	crOwnerRefLabel = "cr-owner.ticdc.tidbcloud.com"

	securityGroupRuleTypeIngress = "ingress"
	securityGroupRuleTypeEgress  = "egress"

	asgResourceType = "auto-scaling-group"
)

var taintMap = map[corev1.TaintEffect]string{
	corev1.TaintEffectNoSchedule:       "NO_SCHEDULE",
	corev1.TaintEffectPreferNoSchedule: "PREFER_NO_SCHEDULE",
	corev1.TaintEffectNoExecute:        "NO_EXECUTE",
}

type aws struct {
	appbox         *regionalpb.APPBox
	assumeRoleName string
	externalID     string
	region         string
	ec2Api         *ec2.EC2
	asgApi         *autoscaling.AutoScaling

	ticdcServiceCfg *config.Config
}

func NewAWSProvider(
	appbox *regionalpb.APPBox,
	ticdcServiceCfg *config.Config,
) Provider {
	externalID := ticdcServiceCfg.AWS.ExternalID
	assumeRoleName := ticdcServiceCfg.AWS.AssumeRoleName
	region := ticdcServiceCfg.Region
	sess := session.Must(session.NewSession())
	cred := stscreds.NewCredentials(sess, util.GetRoleARN(appbox.Kubernetes.AccountId, assumeRoleName),
		func(provider *stscreds.AssumeRoleProvider) {
			provider.ExternalID = aws_sdk.String(externalID)
		})
	ec2Api := ec2.New(sess, &aws_sdk.Config{
		Region:      aws_sdk.String(region),
		Credentials: cred,
	})
	asgApi := autoscaling.New(sess, &aws_sdk.Config{
		Region:      aws_sdk.String(region),
		Credentials: cred,
	})

	return &aws{
		appbox:          appbox,
		assumeRoleName:  assumeRoleName,
		externalID:      externalID,
		ticdcServiceCfg: ticdcServiceCfg,
		region:          region,
		ec2Api:          ec2Api,
		asgApi:          asgApi,
	}
}

func (a *aws) BuildNodePool(providerConfig string, name types.NamespacedName, np *NodePool) (client.Object, error) {
	if len(np.NodeSubnets) < 1 {
		return nil, fmt.Errorf("missing node subnet")
	}
	if a.appbox.Kubernetes.GetAws().TicdcNodeRole == "" {
		return nil, fmt.Errorf("node role is not ready")
	}
	var taints = make([]eksv1beta1.TaintParameters, 0, len(np.Taints))
	for _, t := range np.Taints {
		taint := eksv1beta1.TaintParameters{
			Effect: ptr.To(taintMap[t.Effect]),
			Key:    ptr.To(t.Key),
			Value:  ptr.To(t.Value),
		}
		taints = append(taints, taint)
	}
	if np.Labels == nil {
		np.Labels = make(map[string]string)
	}
	np.Labels[crOwnerRefLabel] = shortName(np.Name)
	ng := &eksv1beta1.NodeGroup{
		ObjectMeta: metav1.ObjectMeta{
			Labels: appendAppboxIDLabel(np.Labels, a.appbox.Name),
			Name:   shortName(np.Name),
		},
		Spec: eksv1beta1.NodeGroupSpec{
			ForProvider: eksv1beta1.NodeGroupParameters{
				ClusterName:   ptr.To(a.appbox.Kubernetes.Name),
				CapacityType:  ptr.To("ON_DEMAND"),
				Region:        ptr.To(np.Region),
				InstanceTypes: []*string{ptr.To(np.InstanceType)},
				Labels:        util.ToMapPointer(np.Labels),
				Tags:          util.ToMapPointer(np.Tags),
				NodeRoleArn:   ptr.To(a.appbox.Kubernetes.GetAws().TicdcNodeRole),
				SubnetIds:     util.ToSlicePointer(np.NodeSubnets),
				ScalingConfig: []eksv1beta1.ScalingConfigParameters{
					{
						MaxSize: ptr.To(float64(maxNodeCount)),
						MinSize: ptr.To(float64(0)),
					},
				},
				AMIType: ptr.To("BOTTLEROCKET_x86_64"),
				Taint:   taints,
			},
			InitProvider: eksv1beta1.NodeGroupInitParameters{
				ScalingConfig: []eksv1beta1.ScalingConfigInitParameters{
					{
						DesiredSize: ptr.To(float64(0)),
					},
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: shortName(providerConfig),
				},
			},
		},
	}
	return ng, nil
}

func (a *aws) PostCreateNodePool(ctx context.Context, cli client.Client, resourceTags map[string]string) error {
	labelsOpt := client.MatchingLabels{
		namer.ComponentSelectorKey: string(cloudv1.ComponentTiCDC),
		ticdc_consts.AppboxIDLabel: a.appbox.Name,
	}
	ngList := &eksv1beta1.NodeGroupList{}
	if err := cli.List(ctx, ngList, labelsOpt); err != nil {
		return fmt.Errorf("list node groups error: %w", err)
	}

	for _, ng := range ngList.Items {
		if len(ng.Annotations) > 0 && ng.Annotations[ticdc_consts.AWSAsgResourceTagsAnnotation] != "" {
			continue
		}

		if !isResourceReady(ng.Status.ResourceStatus) {
			return fmt.Errorf("node group %s is not ready", ng.Name)
		}

		if err := a.addResourceTagsToASG(ctx, cli, &ng, resourceTags); err != nil {
			return fmt.Errorf("add resource tags to asg of node group %s error: %w", ng.Name, err)
		}
	}
	return nil
}

func (a *aws) addResourceTagsToASG(ctx context.Context, cli client.Client, ng *eksv1beta1.NodeGroup, resourceTags map[string]string) error {
	req := &autoscaling.CreateOrUpdateTagsInput{}
	for _, resource := range ng.Status.AtProvider.Resources {
		for _, asg := range resource.AutoscalingGroups {
			if asg.Name == nil || *asg.Name == "" {
				return fmt.Errorf("asg name is empty")
			}
			for k, v := range resourceTags {
				req.Tags = append(req.Tags, &autoscaling.Tag{
					ResourceId:        asg.Name,
					ResourceType:      aws_sdk.String(asgResourceType),
					Key:               aws_sdk.String(k),
					Value:             aws_sdk.String(v),
					PropagateAtLaunch: aws_sdk.Bool(true),
				})
			}
		}
	}
	if len(req.Tags) == 0 {
		// if there is no asg but the node group is ready, we just return nil to not block the ticdc creation
		// and wait next reconcile to try to add resource tags
		log.Warn("no asg found in the node group when adding resource tags", zap.String("node_group_name", ng.Name))
		return nil
	}

	_, err := a.asgApi.CreateOrUpdateTags(req)
	if err != nil {
		return fmt.Errorf("add resource tags to asg error: %w", err)
	}

	if ng.Annotations == nil {
		ng.Annotations = make(map[string]string)
	}
	ng.Annotations[ticdc_consts.AWSAsgResourceTagsAnnotation] = resourceTags2String(resourceTags)
	if err := cli.Update(ctx, ng); err != nil {
		return fmt.Errorf("add resource tag annotation to node group error: %w", err)
	}
	return nil
}

func (a *aws) DeleteNodePool(ctx context.Context, cli client.Client, name types.NamespacedName) error {
	np := eksv1beta1.NodeGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name: shortName(name.Name),
		},
	}
	return k8s.DeleteObject(ctx, cli, &np)
}

func (a *aws) BuildStorageClass(name types.NamespacedName, class *StorageClass) (client.Object, error) {
	params := map[string]string{
		"csi.storage.k8s.io/fstype": "ext4",
		"type":                      "gp3",
		"encrypted":                 "true",
		// Pass the tags to the persistent volumes
		// See https://github.com/kubernetes-sigs/aws-ebs-csi-driver/blob/master/docs/TAGGING.md for more information
		"tagSpecification": fmt.Sprintf("%s=%s", consts.TagKeyCrossplaneVendor, consts.TagValCrossplaneVendor),
	}
	params["iops"] = fmt.Sprint(class.Iops)
	params["throughput"] = fmt.Sprint(class.ThroughputMbs)
	res := &storagev1.StorageClass{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
		VolumeBindingMode:    ptr.To(storagev1.VolumeBindingWaitForFirstConsumer),
		AllowVolumeExpansion: ptr.To(true),
		ReclaimPolicy:        ptr.To(corev1.PersistentVolumeReclaimDelete),
		Provisioner:          "ebs.csi.aws.com",
		Parameters:           params,
		MountOptions:         []string{"nodelalloc,noatime"},
	}
	return res, nil
}

func (a *aws) DeleteStorageClass(ctx context.Context, cli client.Client, name types.NamespacedName) error {
	sc := storagev1.StorageClass{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
	}
	return k8s.DeleteObject(ctx, cli, &sc)
}

func (a *aws) BuildProviderConfig(name types.NamespacedName) (client.Object, error) {
	return &awsv1beta1.ProviderConfig{
		ObjectMeta: metav1.ObjectMeta{
			Name:   shortName(name.Name),
			Labels: appendAppboxIDLabel(nil, a.appbox.Name),
		},
		Spec: awsv1beta1.ProviderConfigSpec{
			Credentials: awsv1beta1.ProviderCredentials{
				Source: "IRSA",
			},
			AssumeRoleChain: []awsv1beta1.AssumeRoleOptions{
				{
					RoleARN:    ptr.To(util.GetRoleARN(a.appbox.Kubernetes.AccountId, a.assumeRoleName)),
					ExternalID: ptr.To(a.externalID),
				},
			},
		},
	}, nil
}

func (a *aws) DeleteProviderConfig(ctx context.Context, cli client.Client, name types.NamespacedName) error {
	pc := awsv1beta1.ProviderConfig{
		ObjectMeta: metav1.ObjectMeta{
			Name: shortName(name.Name),
		},
	}
	return k8s.DeleteObject(ctx, cli, &pc)
}

func (a *aws) GetKubeClient(ctx context.Context) (client.Client, error) {
	opts := client.Options{
		Scheme: scheme.Scheme,
	}
	restCfg, err := a.getConfig()
	if err != nil {
		return nil, err
	}
	return client.New(restCfg, opts)
}

func (a *aws) GetKubernetesVersion(ctx context.Context) (string, error) {
	return a.appbox.Kubernetes.Version, nil
}

func (a *aws) GetKubernetesName(ctx context.Context) (string, error) {
	return a.appbox.Kubernetes.Name, nil
}

func (a *aws) GetKubernetesID(ctx context.Context) (string, error) {
	return a.appbox.Kubernetes.Name, nil
}

func (a *aws) GetNodeZones(ctx context.Context) ([]string, error) {
	var zones = make([]string, 0, len(a.appbox.Kubernetes.GetAws().GetNodeSubnets()))
	for _, subNet := range a.appbox.Kubernetes.GetAws().GetNodeSubnets() {
		zones = append(zones, subNet.GetAzId())
	}
	return zones, nil
}

func (a *aws) GetNodeSubnets(ctx context.Context) ([]string, error) {
	subnets := a.appbox.Kubernetes.GetAws().GetNodeSubnets()
	nets := make([]string, len(subnets))
	for idx, subnet := range subnets {
		nets[idx] = subnet.Id
	}
	return nets, nil
}

func (a *aws) GetPodSubnets(ctx context.Context) ([]string, error) {
	subnets := a.appbox.Kubernetes.GetAws().GetNodeSubnets()
	nets := make([]string, len(subnets))
	for idx, subnet := range subnets {
		nets[idx] = subnet.Id
	}
	return nets, nil
}

func (a *aws) GetSecondaryNodeSubnets(context.Context) ([]string, error) {
	subnets := a.appbox.Kubernetes.GetAws().GetSecondaryNodeSubnets()
	nets := make([]string, len(subnets))
	for idx, subnet := range subnets {
		nets[idx] = subnet.Id
	}
	return nets, nil
}

// GetSinkEndpointDetail gets the private endpoint detail for aws using crossplane
func (a *aws) GetSinkEndpointDetail(ctx context.Context, cli client.Client,
	name types.NamespacedName) (*SinkEndpoint, error) {
	vpce := &ec2v1beta1.VPCEndpoint{}
	if err := cli.Get(ctx, name, vpce); err != nil {
		return nil, errors.Trace(err)
	}
	detail := &SinkEndpoint{}
	detail.Phase = cdcv1.EndpointResourceStatePending
	if vpce.Status.AtProvider.State != nil {
		detail.Phase = awsVPCEndpointStatus2SinkEndpointPhase(*vpce.Status.AtProvider.State)
	}
	if vpce.Status.AtProvider.ID != nil {
		detail.EndpointID = *vpce.Status.AtProvider.ID
	}
	azIDMap := make(map[string]string)
	for _, entry := range a.appbox.Kubernetes.GetAws().GetNodeSubnets() {
		azIDMap[entry.AzName] = entry.AzId
	}
	if len(vpce.Status.AtProvider.DNSEntry) > 0 {
		regional, zoneAddress, err := parsePLEDNSEntries(vpce.Status.AtProvider.DNSEntry, azIDMap)
		if err != nil {
			return nil, errors.Trace(err)
		}
		detail.Address = regional
		detail.ZoneAddress = zoneAddress
	}

	// crossplane take a long time to refresh the status, try to use aws SDK to get the detail ASAP
	if detail.EndpointID != "" && (len(detail.Address) == 0 || len(detail.ZoneAddress) == 0) {
		result, err := a.ec2Api.DescribeVpcEndpoints(&ec2.DescribeVpcEndpointsInput{
			VpcEndpointIds: aws_sdk.StringSlice([]string{detail.EndpointID}),
		})
		if err != nil {
			return nil, errors.Trace(err)
		}
		if len(result.VpcEndpoints) == 0 {
			return nil, fmt.Errorf("VPC endpoint %s not found", detail.EndpointID)
		}
		endpoint := result.VpcEndpoints[0]
		if endpoint.State != nil {
			detail.Phase = awsVPCEndpointStatus2SinkEndpointPhase(*endpoint.State)
		}
		if len(endpoint.DnsEntries) > 0 {
			dnsEntries := make([]ec2v1beta1.DNSEntryObservation, 0, len(endpoint.DnsEntries))
			for _, entry := range endpoint.DnsEntries {
				dnsEntries = append(dnsEntries, ec2v1beta1.DNSEntryObservation{
					DNSName:      entry.DnsName,
					HostedZoneID: entry.HostedZoneId,
				})
			}
			regional, zoneAddress, err := parsePLEDNSEntries(dnsEntries, azIDMap)
			if err != nil {
				return nil, errors.Trace(err)
			}
			detail.Address = regional
			detail.ZoneAddress = zoneAddress
		}
	}
	return detail, nil
}

func (a *aws) BuildSinkEndpoint(providerConfig string,
	name types.NamespacedName,
	endpoint *EndpointRequest) ([]client.Object, error) {
	vpcEndpoint := a.buildVpcEndpoint(name, endpoint, providerConfig)
	securityGroup := a.buildSecurityGroup(vpcEndpoint.Name, endpoint, providerConfig)
	sgIngressRule := a.buildSgRule(securityGroupRuleTypeIngress, securityGroup.Name, endpoint.Region, providerConfig)
	sgEgressRule := a.buildSgRule(securityGroupRuleTypeEgress, securityGroup.Name, endpoint.Region, providerConfig)
	sgAssociation := a.buildSgAssociation(vpcEndpoint.Name, securityGroup.Name, endpoint, providerConfig)
	var objs = []client.Object{
		vpcEndpoint,
		securityGroup,
		sgIngressRule,
		sgEgressRule,
		sgAssociation,
	}

	var nodeSubnets []*regionalpb.APPBox_Kubernetes_AwsSubnet
	if len(a.appbox.Kubernetes.GetAws().GetSecondaryNodeSubnets()) > 0 {
		// if there are secondary node subnets, prefer using that.
		nodeSubnets = a.appbox.Kubernetes.GetAws().GetSecondaryNodeSubnets()
	} else {
		nodeSubnets = a.appbox.Kubernetes.GetAws().GetNodeSubnets()
	}

	azID2SubnetMap, azName2SubnetMap := make(map[string]string), make(map[string]string)
	for _, subnet := range nodeSubnets {
		azID2SubnetMap[subnet.AzId] = subnet.Id
		azName2SubnetMap[subnet.AzName] = subnet.Id
	}
	var (
		subnetIDs []string
		err       error
	)
	if len(endpoint.SelectedZones) > 0 {
		subnetIDs, err = a.getMatchedSubnetsBySelectedZones(endpoint.SelectedZones, azID2SubnetMap)
	} else {
		// for private link to serverless
		subnetIDs, err = a.getMatchedSubnetsByEndpointService(endpoint.EndpointServiceName, azName2SubnetMap)
	}
	if err != nil {
		return nil, err
	}
	for _, subnetID := range subnetIDs {
		subnetAssociation := a.buildVPCEndpointSubnetAssociation(
			subnetID, vpcEndpoint.Name, endpoint.Region, providerConfig)
		objs = append(objs, subnetAssociation)
	}

	return objs, nil
}

func (a *aws) getMatchedSubnetsBySelectedZones(selectedZones []string, az2subnet map[string]string) ([]string, error) {
	subnetIDs := make([]string, 0, len(selectedZones))
	for _, zone := range selectedZones {
		subnetId, ok := az2subnet[zone]
		if !ok {
			return nil, fmt.Errorf("zone %s not found in subnets", zone)
		}
		subnetIDs = append(subnetIDs, subnetId)
	}
	return subnetIDs, nil
}

func (a *aws) getMatchedSubnetsByEndpointService(serviceName string, az2subnet map[string]string) ([]string, error) {
	output, err := a.ec2Api.DescribeVpcEndpointServices(&ec2.DescribeVpcEndpointServicesInput{
		ServiceNames: aws_sdk.StringSlice([]string{serviceName}),
	})
	if err != nil {
		return nil, fmt.Errorf("describe vpc endpoint service %s error: %w", serviceName, err)
	}
	if len(output.ServiceDetails) == 0 {
		return nil, fmt.Errorf("described vpc endpoint service %s is empty", serviceName)
	}

	serviceZones := aws_sdk.StringValueSlice(output.ServiceDetails[0].AvailabilityZones)
	subnetIDs := make([]string, 0, len(serviceZones))
	for _, zone := range serviceZones {
		subnetID, ok := az2subnet[zone]
		if !ok {
			continue
		}
		subnetIDs = append(subnetIDs, subnetID)
	}
	if len(subnetIDs) == 0 {
		return nil, fmt.Errorf(
			"service %s zones %v are not matched with tidb cluster zones %v", serviceName, serviceZones, az2subnet)
	}
	return subnetIDs, nil
}

// DeleteSinkEndpoint deletes a private endpoint for AWS using crossplane
func (a *aws) DeleteSinkEndpoint(ctx context.Context, cli client.Client, name types.NamespacedName) error {
	labelsOpt := client.MatchingLabels{
		crOwnerRefLabel: name.Name,
	}
	if err := cli.DeleteAllOf(ctx, &ec2v1beta1.VPCEndpointSubnetAssociation{}, labelsOpt); client.IgnoreNotFound(err) != nil {
		return errors.Annotate(err, "delete VPCEndpointSubnetAssociations failed")
	}
	if err := k8s.DeleteObject(ctx, cli, &ec2v1beta1.SecurityGroupRule{
		ObjectMeta: metav1.ObjectMeta{
			Name: getSecurityGroupRuleName(securityGroupRuleTypeIngress, name.Name),
		},
	}); err != nil {
		return errors.Annotate(err, "delete ingress SecurityGroupRule failed")
	}
	if err := k8s.DeleteObject(ctx, cli, &ec2v1beta1.SecurityGroupRule{
		ObjectMeta: metav1.ObjectMeta{
			Name: getSecurityGroupRuleName(securityGroupRuleTypeEgress, name.Name),
		},
	}); err != nil {
		return errors.Annotate(err, "delete egress SecurityGroupRule failed")
	}
	if err := k8s.DeleteObject(ctx, cli, &ec2v1beta1.VPCEndpointSecurityGroupAssociation{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
	}); err != nil {
		return errors.Annotate(err, "delete VPCEndpointSecurityGroupAssociation failed")
	}
	if err := k8s.DeleteObject(ctx, cli, &ec2v1beta1.SecurityGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
	}); err != nil {
		return errors.Annotate(err, "delete SecurityGroup failed")
	}
	if err := k8s.DeleteObject(ctx, cli, &ec2v1beta1.VPCEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
		},
	}); err != nil {
		return errors.Annotate(err, "delete VPCEndpoint failed")
	}
	return nil
}

func (a *aws) ValidateEndpointZones(ctx context.Context, endpointZones []string, serviceName string) error {
	tidbZoneID2Name := make(map[string]string)
	for _, subnet := range a.appbox.Kubernetes.GetAws().NodeSubnets {
		tidbZoneID2Name[subnet.AzId] = subnet.AzName
	}
	for _, endpointZoneID := range endpointZones {
		if _, ok := tidbZoneID2Name[endpointZoneID]; !ok {
			return status.Errorf(
				codes.InvalidArgument, "endpoint zone %s is not found in tidbcloud zones", endpointZoneID)
		}
	}

	output, err := a.ec2Api.DescribeVpcEndpointServices(&ec2.DescribeVpcEndpointServicesInput{
		ServiceNames: aws_sdk.StringSlice([]string{serviceName}),
	})
	if err != nil {
		return status.Errorf(
			codes.Internal, "describe vpc endpoint service %s error: %s", serviceName, err.Error())
	}
	if len(output.ServiceDetails) == 0 {
		return status.Errorf(codes.InvalidArgument, "service %s is not found", serviceName)
	}

	serviceZones := aws_sdk.StringValueSlice(output.ServiceDetails[0].AvailabilityZones)
	for _, endpointZoneID := range endpointZones {
		endpointZoneName := tidbZoneID2Name[endpointZoneID]
		if !slices.Contains(serviceZones, endpointZoneName) {
			return status.Errorf(
				codes.InvalidArgument, "endpoint zone %s is not found in service zones", endpointZoneID)
		}
	}
	return nil
}

func (a *aws) BuildDNS(providerConfig string, dns *DNSRequest) ([]client.Object, error) {
	hostedZone := a.buildHostedZone(dns, providerConfig)
	var objs = []client.Object{hostedZone}
	for zone, recordName := range dns.ZoneLevelDNS {
		domainSuffix := fmt.Sprintf(".%s", dns.RootDomainName)
		recordName = strings.TrimSuffix(recordName, domainSuffix)
		address, ok := dns.Endpoint.ZoneAddress[zone]
		if !ok {
			return nil, fmt.Errorf("zone %s not found in zone address, endpoint zone address: %v", zone, dns.Endpoint.ZoneAddress)
		}
		zoneRecord := a.buildZoneRecord(zone, address, fmt.Sprintf("*.%s", recordName), dns, providerConfig)
		objs = append(objs, zoneRecord)
	}
	// add a base record to resolve regional endpoint address
	objs = append(objs, a.buildZoneRecord("all", dns.Endpoint.Address, "*", dns, providerConfig))
	return objs, nil
}

func (a *aws) DeleteDNS(ctx context.Context, cli client.Client, name string) error {
	var records route53v1beta1.RecordList
	// list the resources (zone record) with the label
	err := cli.List(ctx, &records, client.MatchingLabels{
		crOwnerRefLabel: name,
	})
	if err != nil {
		return errors.Annotate(err, "list zone records failed")
	}
	for _, record := range records.Items {
		if err := k8s.DeleteObjectCompletedWithHooks(
			ctx,
			cli,
			&record,
			k8s.HookAutoUnpauseCrossplaneCR,
		); err != nil {
			return errors.Annotate(err, "delete zone record failed")
		}
	}
	if err := k8s.DeleteObjectCompletedWithHooks(ctx, cli, &route53v1beta1.Zone{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}, k8s.HookAutoUnpauseCrossplaneCR); err != nil {
		return errors.Annotate(err, "delete zone failed")
	}
	return nil
}

func (a *aws) getConfig() (*rest.Config, error) {
	if a.ticdcServiceCfg.Debug.Enable && a.ticdcServiceCfg.Debug.DataPlaneKubeConfig != "" {
		return clientcmd.BuildConfigFromFlags("", a.ticdcServiceCfg.Debug.DataPlaneKubeConfig)
	}
	args := []string{
		"token",
		"--region", a.ticdcServiceCfg.Region,
		"--cluster-id", a.appbox.Kubernetes.Name,
	}

	if a.assumeRoleName != "" {
		args = append(args, "--role",
			util.GetRoleARN(a.appbox.Kubernetes.AccountId, a.assumeRoleName))
		if a.externalID != "" {
			args = append(args, "--external-id", a.externalID)
		}
	}
	cfg := clientcmdapi.NewConfig()

	cfgctx := clientcmdapi.NewContext()
	cfgctx.Cluster = a.appbox.Kubernetes.Name
	cfgctx.AuthInfo = a.appbox.Kubernetes.Name

	cluster := clientcmdapi.NewCluster()
	if a.appbox.Kubernetes.Ca != "" {
		cluster.CertificateAuthorityData = []byte(a.appbox.Kubernetes.Ca)
	} else {
		cluster.InsecureSkipTLSVerify = true
	}
	cluster.Server = a.appbox.Kubernetes.Endpoint
	auth := clientcmdapi.NewAuthInfo()
	auth.Exec = &clientcmdapi.ExecConfig{
		APIVersion:      "client.authentication.k8s.io/v1beta1",
		InteractiveMode: clientcmdapi.NeverExecInteractiveMode,
		Command:         "aws-iam-authenticator",
		Args:            args,
	}

	cfg.Contexts[a.appbox.Kubernetes.Name] = cfgctx
	cfg.Clusters[a.appbox.Kubernetes.Name] = cluster
	cfg.AuthInfos[a.appbox.Kubernetes.Name] = auth
	cfg.CurrentContext = a.appbox.Kubernetes.Name
	return clientcmd.NewDefaultClientConfig(*cfg, nil).ClientConfig()
}

func (a *aws) buildVPCEndpointSubnetAssociation(subnetId string,
	vpcEndpointName string,
	region string,
	providerConfig string) *ec2v1beta1.VPCEndpointSubnetAssociation {
	return &ec2v1beta1.VPCEndpointSubnetAssociation{
		ObjectMeta: metav1.ObjectMeta{
			Name: buildVpceSubnetAssociationName(vpcEndpointName, subnetId),
			Labels: map[string]string{
				crOwnerRefLabel: vpcEndpointName,
			},
		},
		Spec: ec2v1beta1.VPCEndpointSubnetAssociationSpec{
			ForProvider: ec2v1beta1.VPCEndpointSubnetAssociationParameters{
				Region:   ptr.To(region),
				SubnetID: ptr.To(subnetId),
				VPCEndpointIDRef: &xpv1.Reference{
					Name: vpcEndpointName,
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: shortName(providerConfig),
				},
			},
		},
	}
}

func (a *aws) buildSecurityGroup(vpcEndpointName string,
	endpoint *EndpointRequest, providerConfig string) *ec2v1beta1.SecurityGroup {
	securityGroup := &ec2v1beta1.SecurityGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name: vpcEndpointName,
			Labels: map[string]string{
				crOwnerRefLabel: vpcEndpointName,
			},
		},
		Spec: ec2v1beta1.SecurityGroupSpec{
			ForProvider: ec2v1beta1.SecurityGroupParameters_2{
				Region:      ptr.To(endpoint.Region),
				Name:        ptr.To(vpcEndpointName),
				Description: ptr.To("allow all security group"),
				VPCID:       ptr.To(a.appbox.Kubernetes.Vpc),
				Tags:        util.ToMapPointer(endpoint.Tags),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: shortName(providerConfig),
				},
			},
		},
	}
	return securityGroup
}

func (a *aws) buildSgAssociation(
	vpcEndpointName string,
	securityGroupName string,
	endpoint *EndpointRequest,
	providerConfig string) *ec2v1beta1.VPCEndpointSecurityGroupAssociation {
	sgAssociation := &ec2v1beta1.VPCEndpointSecurityGroupAssociation{
		ObjectMeta: metav1.ObjectMeta{
			Name: vpcEndpointName,
		},
		Spec: ec2v1beta1.VPCEndpointSecurityGroupAssociationSpec{
			ForProvider: ec2v1beta1.VPCEndpointSecurityGroupAssociationParameters{
				Region: ptr.To(endpoint.Region),
				VPCEndpointIDRef: &xpv1.Reference{
					Name: vpcEndpointName,
				},
				SecurityGroupIDRef: &xpv1.Reference{
					Name: securityGroupName,
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: shortName(providerConfig),
				},
			},
		},
	}
	return sgAssociation
}

func (a *aws) buildVpcEndpoint(name types.NamespacedName,
	endpoint *EndpointRequest, providerConfig string) *ec2v1beta1.VPCEndpoint {
	vpcEndpoint := &ec2v1beta1.VPCEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name: name.Name,
			Labels: map[string]string{
				crOwnerRefLabel: name.Name,
			},
		},
		Spec: ec2v1beta1.VPCEndpointSpec{
			ForProvider: ec2v1beta1.VPCEndpointParameters_2{
				Region:          ptr.To(endpoint.Region),
				ServiceName:     ptr.To(endpoint.EndpointServiceName),
				VPCEndpointType: ptr.To(ec2.VpcEndpointTypeInterface),
				Tags:            util.ToMapPointer(endpoint.Tags),
				VPCID:           ptr.To(a.appbox.Kubernetes.Vpc),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: shortName(providerConfig),
				},
			},
		},
	}
	return vpcEndpoint
}

func (a *aws) buildSgRule(ruleType, sgName, region, providerConfig string) *ec2v1beta1.SecurityGroupRule {
	return &ec2v1beta1.SecurityGroupRule{
		ObjectMeta: metav1.ObjectMeta{
			Name: getSecurityGroupRuleName(ruleType, sgName),
			Labels: map[string]string{
				crOwnerRefLabel: sgName,
			},
		},
		Spec: ec2v1beta1.SecurityGroupRuleSpec{
			ForProvider: ec2v1beta1.SecurityGroupRuleParameters_2{
				Region: ptr.To(region),
				SecurityGroupIDRef: &xpv1.Reference{
					Name: sgName,
				},
				Protocol:    ptr.To("all"),
				Description: ptr.To("allow all"),
				FromPort:    ptr.To(float64(1)),
				ToPort:      ptr.To(float64(65535)),
				Type:        ptr.To(ruleType),
				CidrBlocks: []*string{
					ptr.To("0.0.0.0/0"),
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: shortName(providerConfig),
				},
			},
		},
	}
}

func (a *aws) buildZoneRecord(zone, address,
	recordName string, dns *DNSRequest, providerConfig string) *route53v1beta1.Record {
	return &route53v1beta1.Record{
		ObjectMeta: metav1.ObjectMeta{
			Name: buildDNSRecordName(dns.Name, zone),
			Labels: map[string]string{
				crOwnerRefLabel: dns.Name,
			},
		},
		Spec: route53v1beta1.RecordSpec{
			ForProvider: route53v1beta1.RecordParameters{
				Region: ptr.To(dns.Region),
				Records: []*string{
					ptr.To(address),
				},
				Name: ptr.To(recordName),
				Type: ptr.To("CNAME"),
				ZoneIDRef: &xpv1.Reference{
					Name: dns.Name,
				},
				TTL: ptr.To(float64(300)),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: shortName(providerConfig),
				},
			},
		},
	}
}

func (a *aws) buildHostedZone(dns *DNSRequest, providerConfig string) *route53v1beta1.Zone {
	hostedZone := &route53v1beta1.Zone{
		ObjectMeta: metav1.ObjectMeta{
			Name: dns.Name,
			Labels: map[string]string{
				crOwnerRefLabel: dns.Name,
			},
		},
		Spec: route53v1beta1.ZoneSpec{
			ForProvider: route53v1beta1.ZoneParameters{
				Name: ptr.To(dns.RootDomainName),
				VPC: []route53v1beta1.VPCParameters{
					{
						VPCID: ptr.To(a.appbox.Kubernetes.Vpc),
					},
				},
				Region: ptr.To(dns.Region),
				Tags:   util.ToMapPointer(dns.Tags),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: shortName(providerConfig),
				},
			},
		},
	}
	return hostedZone
}

const (
	pleMiniDNSEntriesNum         = 2
	zonalDNSLevelNum             = 6
	partsOfSixLevelOfZonalDNS    = 6
	partsOfSixLevelOfRegionalDNS = 3
)

// Private Link Endpoint DNS entry example:
//
//			 [{
//				DnsName: "vpce-023af6250979c0365-j9m3ti32.vpce-svc-0e427c5d93acce18f.us-west-2.vpce.amazonaws.com",
//				HostedZoneId: "Z1YSA3EXCYUU9Z"
//				},{
//				DnsName: "vpce-023af6250979c0365-j9m3ti32-us-west-2a.vpce-svc-0e427c5d93acce18f.us-west-2.vpce.amazonaws.com",
//				HostedZoneId: "Z1YSA3EXCYUU9Z"
//				},{
//		        DnsName: "privatelink-19964668.vwiyyv21q7co.clusters.tidb-cloud.com",
//	            HostedZoneId: Z09814683NRQ8JG4GH6UE
//		     }
//			 ...
//
// ]
func parsePLEDNSEntries(entries []ec2v1beta1.DNSEntryObservation,
	logicalToPhysicalZoneMap map[string]string,
) (string, map[string]string, error) {
	if len(entries) < pleMiniDNSEntriesNum {
		return "", nil, fmt.Errorf("dns entries of private link should have at least 2, but got %d",
			len(entries))
	}

	var (
		regionalAddress string
		zonalAddresses  = make(map[string]string)
	)
	for i := 0; i < len(entries); i++ {
		dnsName := *entries[i].DNSName
		if !strings.HasSuffix(dnsName, ".vpce.amazonaws.com") {
			// it's the private dns name, skip
			continue
		}
		levels := strings.Split(dnsName, ".")
		if len(levels) != zonalDNSLevelNum {
			return "", nil, fmt.Errorf(
				"zonal dns name should have 6 level, but %d, the dns name: %s ",
				len(levels), dnsName)
		}
		smallParts := strings.Split(levels[0], "-")
		if len(smallParts) == partsOfSixLevelOfRegionalDNS {
			// it's regional address
			regionalAddress = dnsName
			continue
		}
		if len(smallParts) != partsOfSixLevelOfZonalDNS {
			return "", nil, fmt.Errorf(
				"the sixth level of zonal dns name should have 6 parts, but %d, the sixth level: %s ",
				len(smallParts), levels[0])
		}
		logicalAZ := strings.Join([]string{smallParts[3], smallParts[4], smallParts[5]}, "-")
		zonalAddresses[logicalToPhysicalZoneMap[logicalAZ]] = dnsName
	}
	return regionalAddress, zonalAddresses, nil
}

var statusMap = map[string]cdcv1.SinkEndpointPhase{
	"pendingacceptance": cdcv1.EndpointResourceStatePendingAcceptance,
	"available":         cdcv1.EndpointResourceStateEndpointAvailable,
	"deleting":          cdcv1.EndpointResourceStateDeleting,
	"deleted":           cdcv1.EndpointResourceStateDeleted,
	"rejected":          cdcv1.EndpointResourceStateRejected,
	"failed":            cdcv1.EndpointResourceStateFailed,
}

func awsVPCEndpointStatus2SinkEndpointPhase(status string) cdcv1.SinkEndpointPhase {
	if phase, ok := statusMap[strings.ToLower(status)]; ok {
		return phase
	}
	return cdcv1.EndpointResourceStatePending
}

func getSecurityGroupRuleName(ruleType, name string) string {
	return fmt.Sprintf("%s-%s", ruleType, name)
}

type awsGCProvider struct {
	cli client.Client
}

func NewAwsGCProvider(cli client.Client) GCProvider {
	return &awsGCProvider{
		cli: cli,
	}
}

func (a *awsGCProvider) ListAppbox(ctx context.Context) ([]string, error) {
	var ngs eksv1beta1.NodeGroupList
	listOptions, err := resourceWithLabel(ticdc_consts.AppboxIDLabel)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if err := a.cli.List(ctx, &ngs, listOptions); err != nil {
		return nil, errors.Trace(err)
	}
	var appboxIDs = make(map[string]struct{})
	for _, ng := range ngs.Items {
		if ng.Labels != nil {
			appboxID, ok := ng.Labels[ticdc_consts.AppboxIDLabel]
			if !ok {
				log.Warn("no appbox ID label found", zap.Any("node group", ng))
				continue
			}
			appboxIDs[appboxID] = struct{}{}
		}
	}
	return pie.Keys(appboxIDs), nil
}

func (a *awsGCProvider) OnAppboxDeleting(ctx context.Context, appboxId string) error {
	var ngs eksv1beta1.NodeGroupList
	if err := a.cli.List(ctx, &ngs, client.MatchingLabels{
		ticdc_consts.AppboxIDLabel: appboxId,
	}); err != nil {
		return errors.Trace(err)
	}
	for _, ng := range ngs.Items {
		if err := k8s.DeleteObject(ctx, a.cli, &ng); err != nil {
			log.Warn("delete node group failed",
				zap.Any("node group", ng), zap.Error(err))
			return errors.Trace(err)
		}
		log.Info("node group deleted", zap.Any("node group", ng))
	}
	return nil
}

type awsResourceTagCompensater struct {
	cfg       *config.Config
	appboxCli regionalservice.Client
}

func NewAwsResourceTagCompensater(cfg *config.Config) ResourceTagCompensater {
	return &awsResourceTagCompensater{
		cfg:       cfg,
		appboxCli: regionalservice.NewRegionalServiceClient(cfg.RegionalServiceBaseURL),
	}
}

func (a *awsResourceTagCompensater) CompensateComputeTags(ctx context.Context, cli client.Client) error {
	labelsOpt := client.MatchingLabels{
		namer.ComponentSelectorKey: string(cloudv1.ComponentTiCDC),
	}
	ngList := &eksv1beta1.NodeGroupList{}
	if err := cli.List(ctx, ngList, labelsOpt); err != nil {
		return fmt.Errorf("list node groups error: %w", err)
	}

	appbox2NodeGroups := a.groupNodeGroupByAppbox(ngList)
	var errs error
	for appboxID, nodeGroups := range appbox2NodeGroups {
		if err := a.addResourceTagsByAppbox(ctx, cli, appboxID, nodeGroups); err != nil {
			log.Error("add resource tags to ec2 instances error", zap.String("appbox_id", appboxID), zap.Error(err))
			errs = multierr.Append(errs, err)
			continue
		}
		log.Info("add resource tags to ec2 instances successfully", zap.String("appbox_id", appboxID))
	}
	return errs
}

func (a *awsResourceTagCompensater) groupNodeGroupByAppbox(ngList *eksv1beta1.NodeGroupList) map[string][]*eksv1beta1.NodeGroup {
	appbox2NodeGroups := make(map[string][]*eksv1beta1.NodeGroup)
	for i := range ngList.Items {
		ng := &ngList.Items[i]
		if ng.Labels == nil {
			log.Warn("no label in node group", zap.String("node_group_name", ng.Name))
			continue
		}
		appboxID, ok := ng.Labels[ticdc_consts.AppboxIDLabel]
		if !ok {
			log.Warn("no appbox ID label found in node group", zap.String("node_group_name", ng.Name))
			continue
		}
		appbox2NodeGroups[appboxID] = append(appbox2NodeGroups[appboxID], ng)
	}
	return appbox2NodeGroups
}

func (a *awsResourceTagCompensater) addResourceTagsByAppbox(ctx context.Context, cli client.Client, appboxID string, nodeGroups []*eksv1beta1.NodeGroup) error {
	var asgNamesAddASGTag, asgNamesAddEC2Tag []string
	var ngsAddASGTag, ngsAddEC2Tag []*eksv1beta1.NodeGroup
	for _, ng := range nodeGroups {
		asgNames := a.extractAsgNamesFromNodeGroup(ng)
		if len(asgNames) == 0 {
			log.Warn("no asg found in node group", zap.String("node_group_name", ng.Name))
			continue
		}

		if ng.Annotations == nil || ng.Annotations[ticdc_consts.AWSAsgResourceTagsAnnotation] == "" {
			asgNamesAddASGTag = append(asgNamesAddASGTag, asgNames...)
			ngsAddASGTag = append(ngsAddASGTag, ng)
			asgNamesAddEC2Tag = append(asgNamesAddEC2Tag, asgNames...)
			ngsAddEC2Tag = append(ngsAddEC2Tag, ng)
		} else if ng.Annotations[ticdc_consts.AWSEC2ResourceTagsAnnotation] == "" {
			asgNamesAddEC2Tag = append(asgNamesAddEC2Tag, asgNames...)
			ngsAddEC2Tag = append(ngsAddEC2Tag, ng)
		} else {
			log.Info("already added resource tags to ec2 instances in node group", zap.String("node_group_name", ng.Name))
		}
	}

	if len(asgNamesAddASGTag) == 0 && len(asgNamesAddEC2Tag) == 0 {
		log.Info("no asg found for appbox", zap.String("appbox_id", appboxID))
		return nil
	}

	appbox, err := a.appboxCli.GetAPPBoxByID(ctx, appboxID)
	if err != nil {
		return fmt.Errorf("get appbox %s error: %w", appboxID, err)
	}
	if len(appbox.TenantIds) != 1 {
		return fmt.Errorf("appbox %s has multiple tenants %s", appboxID, strings.Join(appbox.TenantIds, ","))
	}
	if len(appbox.ProjectIds) != 1 {
		return fmt.Errorf("appbox %s has multiple projects %s", appboxID, strings.Join(appbox.ProjectIds, ","))
	}
	tenantID, projectID := appbox.TenantIds[0], appbox.ProjectIds[0]
	ec2Api, asgApi := a.constructAWSApi(appbox.Kubernetes.AccountId)
	tags := a.getResourceTags(tenantID, projectID)

	if len(asgNamesAddASGTag) > 0 {
		if err := a.addASGTags4NodeGroups(ctx, cli, asgApi, ngsAddASGTag, asgNamesAddASGTag, tags); err != nil {
			return fmt.Errorf("add tags to asgs %s error: %w", strings.Join(asgNamesAddASGTag, ","), err)
		}
	}
	if len(asgNamesAddEC2Tag) > 0 {
		if err := a.addEC2Tags4NodeGroups(ctx, cli, asgApi, ec2Api, ngsAddEC2Tag, asgNamesAddEC2Tag, tags); err != nil {
			return fmt.Errorf("add tags to ec2 instances of asgs %s error: %w", strings.Join(asgNamesAddEC2Tag, ","), err)
		}
	}
	return nil
}

func (a *awsResourceTagCompensater) extractAsgNamesFromNodeGroup(ng *eksv1beta1.NodeGroup) []string {
	var asgNames []string
	for _, resource := range ng.Status.AtProvider.Resources {
		for _, asg := range resource.AutoscalingGroups {
			if asg.Name == nil || *asg.Name == "" {
				continue
			}
			asgNames = append(asgNames, *asg.Name)
		}
	}
	return asgNames
}

func (a *awsResourceTagCompensater) addEC2Tags4NodeGroups(
	ctx context.Context, cli client.Client, asgApi *autoscaling.AutoScaling, ec2Api *ec2.EC2,
	nodeGroups []*eksv1beta1.NodeGroup, asgNames []string, tags map[string]string,
) error {
	ec2InstanceIDs, err := a.listEC2InstancesByASGNames(asgApi, asgNames)
	if err != nil {
		return fmt.Errorf("list ec2 instances by asg names error: %w", err)
	}
	if len(ec2InstanceIDs) > 0 {
		if err := a.addResourceTags2EC2Instances(ec2Api, ec2InstanceIDs, tags); err != nil {
			return fmt.Errorf("add resource tags %s to ec2 instances %s error: %w", resourceTags2String(tags), strings.Join(ec2InstanceIDs, ","), err)
		}
	}
	if err := a.markEC2ResourceTagsAdded(ctx, cli, nodeGroups, ticdc_consts.AWSEC2ResourceTagsAnnotation, tags); err != nil {
		return fmt.Errorf("mark ec2 resource tags added for node groups error: %w", err)
	}
	return nil
}

func (a *awsResourceTagCompensater) addASGTags4NodeGroups(
	ctx context.Context, cli client.Client, asgApi *autoscaling.AutoScaling,
	nodeGroups []*eksv1beta1.NodeGroup, asgNames []string, tags map[string]string,
) error {
	if err := a.addResourceTags2ASGs(asgApi, asgNames, tags); err != nil {
		return fmt.Errorf("add resource tags %s to asgs error: %w", resourceTags2String(tags), err)
	}
	if err := a.markEC2ResourceTagsAdded(ctx, cli, nodeGroups, ticdc_consts.AWSAsgResourceTagsAnnotation, tags); err != nil {
		return fmt.Errorf("mark asg resource tags added for node groups error: %w", err)
	}
	return nil
}

func (a *awsResourceTagCompensater) listEC2InstancesByASGNames(asgApi *autoscaling.AutoScaling, asgNames []string) ([]string, error) {
	var instanceIDs []string
	step := 100
	for left := 0; left < len(asgNames); {
		right := left + step
		if right > len(asgNames) {
			right = len(asgNames)
		}
		batchASGs := asgNames[left:right]
		req := &autoscaling.DescribeAutoScalingGroupsInput{
			AutoScalingGroupNames: aws_sdk.StringSlice(batchASGs),
			MaxRecords:            aws_sdk.Int64(int64(step)),
		}
		resp, err := asgApi.DescribeAutoScalingGroups(req)
		if err != nil {
			return nil, fmt.Errorf("describe auto scaling groups error: %w", err)
		}

		for _, asg := range resp.AutoScalingGroups {
			for _, instance := range asg.Instances {
				if instance.InstanceId == nil || *instance.InstanceId == "" {
					continue
				}
				instanceIDs = append(instanceIDs, *instance.InstanceId)
			}
		}
		left = right
	}
	return instanceIDs, nil
}

func (a *awsResourceTagCompensater) addResourceTags2EC2Instances(ec2Api *ec2.EC2, instanceIDs []string, tags map[string]string) error {
	ec2Tags := make([]*ec2.Tag, 0, len(tags))
	for k, v := range tags {
		ec2Tags = append(ec2Tags, &ec2.Tag{
			Key:   aws_sdk.String(k),
			Value: aws_sdk.String(v),
		})
	}

	req := &ec2.CreateTagsInput{
		Resources: aws_sdk.StringSlice(instanceIDs),
		Tags:      ec2Tags,
	}
	_, err := ec2Api.CreateTags(req)
	return err
}

func (a *awsResourceTagCompensater) addResourceTags2ASGs(asgApi *autoscaling.AutoScaling, asgNames []string, tags map[string]string) error {
	req := &autoscaling.CreateOrUpdateTagsInput{}
	for _, asgName := range asgNames {
		for k, v := range tags {
			req.Tags = append(req.Tags, &autoscaling.Tag{
				ResourceId:        aws_sdk.String(asgName),
				ResourceType:      aws_sdk.String(asgResourceType),
				Key:               aws_sdk.String(k),
				Value:             aws_sdk.String(v),
				PropagateAtLaunch: aws_sdk.Bool(true),
			})
		}
	}

	_, err := asgApi.CreateOrUpdateTags(req)
	if err != nil {
		return fmt.Errorf("add resource tags to asg error: %w", err)
	}
	return nil
}

func (a *awsResourceTagCompensater) markEC2ResourceTagsAdded(ctx context.Context, cli client.Client, nodeGroups []*eksv1beta1.NodeGroup, annoKey string, tags map[string]string) error {
	var err error
	annoVal := resourceTags2String(tags)
	for _, ng := range nodeGroups {
		if ng.Annotations == nil {
			ng.Annotations = make(map[string]string)
		}
		ng.Annotations[annoKey] = annoVal
		if updateErr := cli.Update(ctx, ng); updateErr != nil {
			err = multierr.Append(err, fmt.Errorf("add annotation to node group %s error: %w", ng.Name, updateErr))
			continue
		}
	}
	return err
}

func (a *awsResourceTagCompensater) constructAWSApi(awsAccountID string) (*ec2.EC2, *autoscaling.AutoScaling) {
	externalID := a.cfg.AWS.ExternalID
	assumeRoleName := a.cfg.AWS.AssumeRoleName
	region := a.cfg.Region
	sess := session.Must(session.NewSession())
	cred := stscreds.NewCredentials(sess, util.GetRoleARN(awsAccountID, assumeRoleName),
		func(provider *stscreds.AssumeRoleProvider) {
			provider.ExternalID = aws_sdk.String(externalID)
		})
	ec2Api := ec2.New(sess, &aws_sdk.Config{
		Region:      aws_sdk.String(region),
		Credentials: cred,
	})
	asgApi := autoscaling.New(sess, &aws_sdk.Config{
		Region:      aws_sdk.String(region),
		Credentials: cred,
	})
	return ec2Api, asgApi
}

func (a *awsResourceTagCompensater) getResourceTags(tenantID, projectID string) map[string]string {
	return map[string]string{
		consts.TagKeyTenant:      tenantID,
		consts.TagKeyProject:     projectID,
		consts.TagKeyUsedBy:      consts.UsedByCustomerTiDB,
		consts.TagKeyServiceType: consts.ServiceTypeDedicated,
		consts.TagKeyEnvironment: a.cfg.Env,
		consts.TagKeyComponent:   string(cloudv1.ComponentTiCDC),
	}
}

func isResourceReady(status xpv1.ResourceStatus) bool {
	condition := status.GetCondition(xpv1.TypeReady)
	return condition.Status == corev1.ConditionTrue && condition.Reason == xpv1.ReasonAvailable
}

func buildVpceSubnetAssociationName(vpceName, subnetID string) string {
	return fmt.Sprintf("%s-%s", vpceName, subnetID)
}

func buildDNSRecordName(hostedZoneName, azName string) string {
	return fmt.Sprintf("%s-%s", hostedZoneName, azName)
}

func resourceTags2String(resourceTags map[string]string) string {
	tagKVs := make([]string, 0, len(resourceTags))
	for k, v := range resourceTags {
		tagKVs = append(tagKVs, fmt.Sprintf("%s:%s", k, v))
	}
	sort.Strings(tagKVs)
	return strings.Join(tagKVs, ",")
}