package cluster

import (
	"context"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/ec2"
	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	"github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi"
	rgtatypes "github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi/types"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/crossplane/crossplane-runtime/pkg/meta"
	ec2v1alpha1 "github.com/crossplane/provider-aws/apis/ec2/v1alpha1"
	route53v1alpha1 "github.com/crossplane/provider-aws/apis/route53/v1alpha1"
	"github.com/pingcap/errors"
	"go.uber.org/multierr"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/controllers/common"
	reconcilerr "github.com/tidbcloud/infra-api/controllers/common/error"
	"github.com/tidbcloud/infra-api/pkg/metrics"
	tracerutil "github.com/tidbcloud/infra-api/pkg/tracer"
	"github.com/tidbcloud/infra-api/pkg/util/labels"
)

const (
	defaultDNSTTLSeconds = 300

	awsEmptyConnectionListWaitingDuration = 5 * time.Minute
)

const (
	LabelKeyAWSLoadBalancerResource = "service.k8s.aws/resource"
	LabelKeyAWSLoadBalancerStack    = "service.k8s.aws/stack"
	LabelKeyAWSLoadBalancerCluster  = "elbv2.k8s.aws/cluster"

	LabelValAWSLoadBalancerResource = "LoadBalancer"

	AnnotationAWSEmptyListWaitingStarted      = "privatelink.cloud.pingcap.com/waiting-started"
	AnnotationAWSEmptyListWaitingStartingTime = "privatelink.cloud.pingcap.com/waiting-starting-time"

	AnnotationCrossplaneCreationPending = "crossplane.io/external-create-pending"
)

const (
	crossplaneProviderConfigReferenceName = "provider-aws"
)

const (
	awsEndpointNotFoundErrorCode = "InvalidVpcEndpoint.NotFound"
)

var (
	checkEndpointPeriodicallyErr error = errors.New("Check endpoint connection status periodically")
)

// TODO(zlj): conversion of ec2 types state to lower case string can be removed after this fix: https://github.com/aws/aws-sdk-go-v2/issues/1730
type awsEC2State string

var (
	awsEC2StatePendingAcceptance awsEC2State = "pendingacceptance"
	awsEC2StatePending           awsEC2State = "pending"
	awsEC2StateAvailable         awsEC2State = "available"
	awsEC2StateDeleting          awsEC2State = "deleting"
	awsEC2StateDeleted           awsEC2State = "deleted"
	awsEC2StateRejected          awsEC2State = "rejected"
	awsEC2StateFailed            awsEC2State = "failed"
	awsEC2StateExpired           awsEC2State = "expired"
)

func newAWSEC2State(e ec2types.State) awsEC2State {
	return awsEC2State(strings.ToLower(string(e)))
}

func (r *reconcileContext) isCheckEndpointPeriodicallyErr(err error) bool {
	return err.Error() == checkEndpointPeriodicallyErr.Error()
}

func (r *reconcileContext) attachWaitingStartingTime(ctx context.Context) (bool, error) {
	err := r.ensureAnnotations(func() error {
		r.Target.Annotations[AnnotationAWSEmptyListWaitingStarted] = "true"
		r.Target.Annotations[AnnotationAWSEmptyListWaitingStartingTime] = time.Now().Format(time.RFC3339)
		return nil
	})
	return false, err
}

func (r *reconcileContext) clearWaitingStarted(ctx context.Context) (bool, error) {
	err := r.ensureAnnotations(func() error {
		r.Target.Annotations[AnnotationAWSEmptyListWaitingStarted] = "false"
		return nil
	})
	return true, err
}

func (r *reconcileContext) checkAWSPrivateLinkConnectionList(ctx context.Context, length int) (bool, error) {
	if length > 0 {
		return r.clearWaitingStarted(ctx)
	}

	started, ok := r.Target.Annotations[AnnotationAWSEmptyListWaitingStarted]
	if !ok || started == "false" {
		return r.attachWaitingStartingTime(ctx)
	}
	// ok and stated == "true"
	startingTime, ok := r.Target.Annotations[AnnotationAWSEmptyListWaitingStartingTime]
	if !ok {
		// Auto fix
		return r.attachWaitingStartingTime(ctx)
	}

	t, err := time.Parse(time.RFC3339, startingTime)
	if err != nil {
		// Auto fix
		return r.attachWaitingStartingTime(ctx)
	}

	if time.Since(t) > awsEmptyConnectionListWaitingDuration {
		return r.clearWaitingStarted(ctx)
	}

	// In waiting duration
	return false, nil
}

func (r *reconcileContext) getPrivateNetworkLoadBalancerARN(ctx context.Context) (string, error) {
	c := r.AWS.ResourceGroupsTaggingAPI()
	res, err := c.GetResources(ctx, &resourcegroupstaggingapi.GetResourcesInput{
		ResourceTypeFilters: []string{
			"elasticloadbalancing:loadbalancer",
		},
		TagFilters: []rgtatypes.TagFilter{
			{
				Key: aws.String(common.GetKubeClusterTagKey(r.shoot.Status.TechnicalID)),
				Values: []string{
					"owned",
				},
			},
			{
				Key: aws.String(common.ServiceName),
				Values: []string{
					fmt.Sprintf("%s/%s", r.tc.Namespace, getPrivateServiceName(r.tc.Name)),
				},
			},
		},
	}, func(opts *resourcegroupstaggingapi.Options) {
		opts.Region = r.Network.Spec.Region
	})
	if err != nil {
		return "", err
	}
	arns := []string{}
	for _, item := range res.ResourceTagMappingList {
		arns = append(arns, *item.ResourceARN)
	}
	if len(arns) == 0 {
		return "", fmt.Errorf("no loadbalancer")
	}
	if len(arns) != 1 {
		return "", fmt.Errorf("only one loadbalancer can be listed, but current has %d: %v", len(arns), arns)
	}

	return arns[0], nil
}

func getPrivateLinkServiceName(ns, clusterName string) string {
	return fmt.Sprintf("%s.%s", ns, clusterName)
}

func (r *reconcileContext) reconcileCrossplaneVPCEndpointServiceConfiguration(ctx context.Context) (*ec2v1alpha1.VPCEndpointServiceConfiguration, error) {

	getAWSPrivateLinkTagSpecification := func() []*ec2v1alpha1.TagSpecification {
		tags := common.ConcatenateTags(cloudv1.TagsFromLabels(r.Target.Labels),
			map[string]string{
				common.Cluster:   r.Target.Name,
				common.Component: common.Loadbalance,
				common.UsedBy:    common.CustomerTiDB,
			})

		ntags := []*ec2v1alpha1.Tag{}

		for key, val := range tags {
			ntags = append(ntags, &ec2v1alpha1.Tag{
				Key:   pointer.String(key),
				Value: pointer.String(val),
			})
		}

		return []*ec2v1alpha1.TagSpecification{
			{
				ResourceType: pointer.String("vpc-endpoint-service"),
				Tags:         ntags,
			},
		}
	}

	vesc := &ec2v1alpha1.VPCEndpointServiceConfiguration{
		ObjectMeta: metav1.ObjectMeta{
			Name: getPrivateLinkServiceName(r.tc.Namespace, r.tc.Name),
		},
	}
	err := r.Client.Get(ctx, client.ObjectKeyFromObject(vesc), vesc)
	if err != nil {
		if !apierrors.IsNotFound(err) {
			return nil, err
		}

		// Create vesc
		nlbARN, err := r.getPrivateNetworkLoadBalancerARN(ctx)
		if err != nil {
			return nil, err
		}

		privateDNSName := getPrivateLinkEndpointAddress(r.publicEndpointAddress())
		vesc.ObjectMeta.Labels = labels.NewOwnerLabels(r.Target, nil)
		vesc.Spec = ec2v1alpha1.VPCEndpointServiceConfigurationSpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: crossplaneProviderConfigReferenceName,
				},
			},
			ForProvider: ec2v1alpha1.VPCEndpointServiceConfigurationParameters{
				Region:             r.Network.Spec.Region,
				AcceptanceRequired: pointer.BoolPtr(true),
				PrivateDNSName:     &privateDNSName,
				CustomVPCEndpointServiceConfigurationParameters: ec2v1alpha1.CustomVPCEndpointServiceConfigurationParameters{
					NetworkLoadBalancerARNs: []*string{
						&nlbARN,
					},
				},
				TagSpecifications: getAWSPrivateLinkTagSpecification(),
			},
		}
		err = r.Client.Create(ctx, vesc)
		if err != nil {
			return nil, err
		}
	}

	sc := vesc.Status.AtProvider.ServiceConfiguration
	if sc == nil || sc.ServiceID == nil {
		return nil, &reconcilerr.RequeueAfterError{
			RequeueAfter: defaultWaitInterval,
			Caused:       errors.New("wait crossplane vpc endpoint service sync"),
		}
	}
	return vesc, nil
}

// reconcilePrivateLinkService reconcile the privatelink service to active
// The returns contain 3 cases:
// 1. nil, nil: No more reconcile
// 2. nil, error: An error was encountered during reconcile
// 3. error, nil: A normal requeue to check endpoint periodically, and no actual error was encountered
func (r *reconcileContext) reconcilePrivateLinkService(ctx context.Context) (_ error, errResult error) {
	spanCtx, span := tracer.Start(ctx, "reconcilePrivateLinkService")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	if err := r.refreshTidbClusterStatus(); err != nil {
		return nil, err
	}

	refreshStatusMessage := func(err error) error {
		message := ""
		if err != nil {
			message = err.Error()
		}
		errEnsure := r.ensureStatus(func() error {
			r.Target.Status.SqlEndpoints.PrivateLinkService.Message = message
			return nil
		})
		return multierr.Append(err, errEnsure)
	}

	// Delete service
	if r.Target.Spec.SqlEndpoints.PrivateLinkService == nil {
		err := r.deletePrivateLinkService(spanCtx)
		if err != nil {
			return nil, refreshStatusMessage(err)
		}
		return nil, nil
	}

	if r.Target.Status.SqlEndpoints.PrivateLinkService == nil {
		err := r.ensureStatus(func() error {
			r.Target.Status.SqlEndpoints.PrivateLinkService = &cloudv1.PrivateLinkServiceStatus{
				Phase:             cloudv1.PrivateLinkServicePhaseCreating,
				AWS:               &cloudv1.AWSPrivateLinkServiceStatus{},
				CreationTimestamp: metav1.NewTime(time.Now()),
			}
			return nil
		})
		if err != nil {
			return nil, err
		}
	}

	status := r.Target.Status.SqlEndpoints.PrivateLinkService.DeepCopy()
	if status.Phase == cloudv1.PrivateLinkServicePhaseCreating {
		metrics.InfraClusterPrivateLinkServiceCreationDuration.WithLabelValues(
			r.Target.Name, r.Target.Namespace,
		).Set(time.Since(status.CreationTimestamp.Time).Seconds())
	}

	// TODO(zlj): delete when endpoints ready in metadb and worker in central
	// Show endpoints in status early
	err := r.ensureStatus(func() error {
		connsMap := r.getConnsMap()
		statusConns := []*cloudv1.AWSPrivateLinkServiceConnection{}
		for id, c := range connsMap {
			statusConns = append(statusConns, &cloudv1.AWSPrivateLinkServiceConnection{
				Endpoint: id,
				Phase:    c.Phase,
				Message:  c.Message,
			})
		}
		// Sort with endpoint ID
		sort.Slice(statusConns, func(i, j int) bool {
			return statusConns[i].Endpoint < statusConns[j].Endpoint
		})
		r.Target.Status.SqlEndpoints.PrivateLinkService.AWS.Connections = statusConns
		return nil
	})
	if err != nil {
		return nil, err
	}

	vesc, err := r.reconcileCrossplaneVPCEndpointServiceConfiguration(spanCtx)
	if err != nil {
		return nil, refreshStatusMessage(err)
	}

	sc := vesc.Status.AtProvider.ServiceConfiguration
	if sc == nil || sc.ServiceID == nil {
		return nil, &reconcilerr.RequeueAfterError{
			RequeueAfter: defaultWaitInterval,
			Caused:       errors.New("wait vpc endpoint service sync"),
		}
	}

	if err := r.reconcilePrivateLinkServicePirvateDNSName(spanCtx, sc); err != nil {
		return nil, refreshStatusMessage(err)
	}

	if sc.ServiceState == nil ||
		*sc.ServiceState != string(ec2types.ServiceStateAvailable) {
		return nil, &reconcilerr.RequeueAfterError{
			RequeueAfter: defaultWaitInterval,
			Caused:       errors.New("wait vpc endpoint service available"),
		}
	}

	if err := r.reconcilePrivateLinkServiceAllowedPrincipals(spanCtx, sc); err != nil {
		return nil, refreshStatusMessage(err)
	}

	// PirvateLink Service is available now
	err = r.ensureStatus(func() error {
		// Separate updates to prevent involving other domains
		r.Target.Status.SqlEndpoints.PrivateLinkService.Phase = cloudv1.PrivateLinkServicePhaseActive
		r.Target.Status.SqlEndpoints.PrivateLinkService.AWS = &cloudv1.AWSPrivateLinkServiceStatus{
			NLB:            *vesc.Spec.ForProvider.CustomVPCEndpointServiceConfigurationParameters.NetworkLoadBalancerARNs[0],
			ServiceID:      *sc.ServiceID,
			ServiceName:    *sc.ServiceName,
			PrivateDNSName: *vesc.Spec.ForProvider.PrivateDNSName,
			Port:           pointer.Int32(r.Target.GetPrivateSQLEndpointPort()),
			Connections:    r.Target.Status.SqlEndpoints.PrivateLinkService.AWS.Connections,
		}
		r.Target.Status.SqlEndpoints.PrivateLinkService.Message = ""
		return nil
	})
	if err != nil {
		return nil, err
	}

	statusConns, err := r.reconcilePrivateLinkServiceConnections(spanCtx, sc)
	if err != nil && !r.isCheckEndpointPeriodicallyErr(err) {
		return nil, refreshStatusMessage(err)
	}

	err = r.ensureStatus(func() error {
		r.Target.Status.SqlEndpoints.PrivateLinkService.AWS.Connections = statusConns
		return nil
	})
	if err != nil {
		return nil, err
	}

	// Check endpoint connections periodically
	return &reconcilerr.RequeueAfterError{
		RequeueAfter: 2 * defaultWaitInterval,
		Caused:       checkEndpointPeriodicallyErr,
	}, nil
}

func (r *reconcileContext) reconcilePrivateLinkServicePirvateDNSName(
	ctx context.Context, sc *ec2v1alpha1.ServiceConfiguration,
) error {
	vescID := *sc.ServiceID
	dnsConfig := sc.PrivateDNSNameConfiguration

	if *dnsConfig.State == string(ec2types.DnsNameStateVerified) {
		return nil
	}

	rrs := &route53v1alpha1.ResourceRecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name: r.privatelinkDnsRecordName(),
		},
	}
	if err := r.Client.Get(ctx, client.ObjectKeyFromObject(rrs), rrs); err != nil {
		if !apierrors.IsNotFound(err) {
			return err
		}
		rrs.ObjectMeta.Annotations = map[string]string{
			meta.AnnotationKeyExternalName: *dnsConfig.Name + "." + r.HostedZoneDomain(),
		}
		rrs.Spec = route53v1alpha1.ResourceRecordSetSpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: crossplaneProviderConfigReferenceName,
				},
			},
			ForProvider: route53v1alpha1.ResourceRecordSetParameters{
				Type: *dnsConfig.Type,
				TTL:  pointer.Int64(defaultDNSTTLSeconds),
				ResourceRecords: []route53v1alpha1.ResourceRecord{
					{
						// TXT record value should be enclosed in quotation marks, for example, "Sample text entry".
						Value: strconv.Quote(*dnsConfig.Value),
					},
				},
				ZoneID: pointer.StringPtr(r.HostedZoneID()),
			},
		}
		err := r.Client.Create(ctx, rrs)
		if err != nil {
			return err
		}
	}
	if rrs.Status.GetCondition(xpv1.TypeSynced).Reason != xpv1.ReasonReconcileSuccess {
		return &reconcilerr.RequeueAfterError{
			RequeueAfter: defaultWaitInterval,
			Caused:       fmt.Errorf("wait route53 ResourceRecordSet condition to %s, condition: %v", xpv1.ReasonReconcileSuccess, rrs.Status.GetCondition(xpv1.TypeSynced)),
		}
	}
	if rrs.Status.GetCondition(xpv1.TypeReady).Reason != xpv1.ReasonAvailable {
		return &reconcilerr.RequeueAfterError{
			RequeueAfter: defaultWaitInterval,
			Caused:       fmt.Errorf("wait route53 ResourceRecordSet condition to %s, condition: %v", xpv1.ReasonAvailable, rrs.Status.GetCondition(xpv1.TypeReady)),
		}
	}

	// Start domain ownership verification
	verifyOpRes, err := r.AWS.Ec2().StartVpcEndpointServicePrivateDnsVerification(ctx, &ec2.StartVpcEndpointServicePrivateDnsVerificationInput{
		ServiceId: &vescID,
	}, func(opts *ec2.Options) {
		opts.Region = r.Network.Spec.Region
	})
	if err != nil {
		return err
	}
	if *verifyOpRes.ReturnValue == false {
		return fmt.Errorf("verify domain ownership failed, dns config: %v", dnsConfig)
	}

	if *dnsConfig.State != string(ec2types.DnsNameStateVerified) {
		return &reconcilerr.RequeueAfterError{
			RequeueAfter: defaultWaitInterval,
			Caused:       fmt.Errorf("wait vpc endpoint service dns verified, current state is %v", *dnsConfig.State),
		}
	}

	return nil
}

func (r *reconcileContext) reconcilePrivateLinkServiceAllowedPrincipals(
	ctx context.Context, sc *ec2v1alpha1.ServiceConfiguration,
) error {
	mOpRes, err := r.AWS.Ec2().ModifyVpcEndpointServicePermissions(ctx, &ec2.ModifyVpcEndpointServicePermissionsInput{
		ServiceId:            sc.ServiceID,
		AddAllowedPrincipals: []string{"*"},
	}, func(opts *ec2.Options) {
		opts.Region = r.Network.Spec.Region
	})
	if err != nil {
		return err
	}
	if mOpRes.ReturnValue == nil || !*mOpRes.ReturnValue {
		return fmt.Errorf("modify vpc endpoint service permission to '*' failed")
	}

	return nil
}

func (r *reconcileContext) getConnsMap() map[string]*cloudv1.AWSPrivateLinkServiceConnection {
	spec := r.Target.Spec.SqlEndpoints.PrivateLinkService.DeepCopy()
	status := r.Target.Status.SqlEndpoints.PrivateLinkService.DeepCopy()

	// Initialize connetions map
	// Inherit from status, and contains all connections
	connsMap := make(map[string]*cloudv1.AWSPrivateLinkServiceConnection)
	for _, c := range status.AWS.Connections {
		connsMap[c.Endpoint] = c
	}

	acceptedIDs := make(map[string]struct{})
	for _, c := range spec.AWS.AcceptedEndpoints {
		acceptedIDs[c] = struct{}{}
		// Add new endpoints to Pending phase
		if _, ok := connsMap[c]; !ok {
			connsMap[c] = &cloudv1.AWSPrivateLinkServiceConnection{
				Endpoint: c,
				Phase:    cloudv1.PrivateLinkServiceConnectionPhasePending,
			}
		}
	}
	isAccepted := func(id string) bool {
		_, ok := acceptedIDs[id]
		return ok
	}
	for _, c := range connsMap {
		if !isAccepted(c.Endpoint) &&
			c.Phase != cloudv1.PrivateLinkServiceConnectionPhaseDeleting {
			c.Phase = cloudv1.PrivateLinkServiceConnectionPhaseDeleting
			c.Message = ""
		}
	}
	return connsMap
}

func (r *reconcileContext) reconcilePrivateLinkServiceConnections(
	ctx context.Context,
	sc *ec2v1alpha1.ServiceConfiguration,
) ([]*cloudv1.AWSPrivateLinkServiceConnection, error) {
	connsMap := r.getConnsMap()

	endpointsToReject := make([]string, 0)
	endpointsToAccept := make([]string, 0)
	for _, c := range connsMap {
		if c.Phase == cloudv1.PrivateLinkServiceConnectionPhaseDeleting {
			endpointsToReject = append(endpointsToReject, c.Endpoint)
		} else if c.Phase != cloudv1.PrivateLinkServiceConnectionPhaseActive {
			endpointsToAccept = append(endpointsToAccept, c.Endpoint)
		}
	}

	if len(endpointsToAccept) > 0 {
		acceptOpRes, err := r.AWS.Ec2().AcceptVpcEndpointConnections(ctx, &ec2.AcceptVpcEndpointConnectionsInput{
			ServiceId:      sc.ServiceID,
			VpcEndpointIds: endpointsToAccept,
		}, func(opts *ec2.Options) {
			opts.Region = r.Network.Spec.Region
		})
		if err != nil {
			return nil, err
		}

		if acceptOpRes.Unsuccessful != nil {
			// Modify connections information
			for _, c := range acceptOpRes.Unsuccessful {
				endpointID := *c.ResourceId
				connsMap[endpointID].Phase = cloudv1.PrivateLinkServiceConnectionPhaseFailed
				connsMap[endpointID].Message = *c.Error.Message
			}
		}
	}

	if len(endpointsToReject) > 0 {
		rejectOpRes, err := r.AWS.Ec2().RejectVpcEndpointConnections(ctx, &ec2.RejectVpcEndpointConnectionsInput{
			ServiceId:      sc.ServiceID,
			VpcEndpointIds: endpointsToReject,
		}, func(opts *ec2.Options) {
			opts.Region = r.Network.Spec.Region
		})
		if err != nil {
			return nil, err
		}
		if rejectOpRes.Unsuccessful != nil {
			// Modify connections information
			for _, c := range rejectOpRes.Unsuccessful {
				endpointID := *c.ResourceId

				// Skip not found endpoint
				if *c.Error.Code == awsEndpointNotFoundErrorCode {
					continue
				}

				// Keep in deleting
				connsMap[endpointID].Phase = cloudv1.PrivateLinkServiceConnectionPhaseDeleting
				connsMap[endpointID].Message = *c.Error.Message
			}
		}
	}

	currentAWSConnectionsMap, err := r.getAWSConnectionsMap(ctx, *sc.ServiceID)
	if err != nil {
		return nil, err
	}
	// AWS will return an empty connection list when some endpoint in deleting state in custom's side
	// Hold and wait for a fixed period to check whether it is real empty
	checkPassed, err := r.checkAWSPrivateLinkConnectionList(ctx, len(currentAWSConnectionsMap))
	if err != nil {
		return nil, err
	}

	// Construct connections with connection map
	statusConns := []*cloudv1.AWSPrivateLinkServiceConnection{}
	for id, c := range connsMap {
		if checkPassed {
			c = r.refreshStatusConnsWithAWS(c, currentAWSConnectionsMap)
		}
		if c == nil {
			continue
		}

		statusConns = append(statusConns, &cloudv1.AWSPrivateLinkServiceConnection{
			Endpoint: id,
			Phase:    c.Phase,
			Message:  c.Message,
		})
	}

	// Sort with endpoint ID
	sort.Slice(statusConns, func(i, j int) bool {
		return statusConns[i].Endpoint < statusConns[j].Endpoint
	})

	// Reconcile again to get real-time status
	return statusConns, checkEndpointPeriodicallyErr
}

func (r *reconcileContext) refreshStatusConnsWithAWS(c *cloudv1.AWSPrivateLinkServiceConnection, m map[string]awsEC2State) *cloudv1.AWSPrivateLinkServiceConnection {
	id := c.Endpoint
	currentState, ok := m[id]
	if ok {
		switch currentState {
		case awsEC2StateAvailable:
			c.Phase = cloudv1.PrivateLinkServiceConnectionPhaseActive
			c.Message = ""
		case awsEC2StateDeleted, awsEC2StateRejected:
			// Deleted
			if c.Phase == cloudv1.PrivateLinkServiceConnectionPhaseDeleting {
				return nil
			}
		case awsEC2StatePending, awsEC2StatePendingAcceptance:
			c.Phase = cloudv1.PrivateLinkServiceConnectionPhasePending
			c.Message = ""
		case awsEC2StateDeleting, awsEC2StateFailed, awsEC2StateExpired:
			// Ignore
		default:
			// Ignore
		}
	} else {
		// Deleted
		if c.Phase == cloudv1.PrivateLinkServiceConnectionPhaseDeleting {
			return nil
		} else if c.Phase == cloudv1.PrivateLinkServiceConnectionPhaseActive ||
			c.Phase == cloudv1.PrivateLinkServiceConnectionPhasePending {
			// The customer deletes the endpoint by himself
			c.Phase = cloudv1.PrivateLinkServiceConnectionPhaseFailed
			c.Message = fmt.Sprintf("The Vpc Endpoint Id '%s' does not exist", id)
		}
	}
	return c
}

func (r *reconcileContext) getAWSConnectionsMap(ctx context.Context, serviceID string) (map[string]awsEC2State, error) {
	connectionsMap := make(map[string]awsEC2State, 0)
	res, err := r.AWS.Ec2().DescribeVpcEndpointConnections(ctx, &ec2.DescribeVpcEndpointConnectionsInput{
		Filters: []ec2types.Filter{
			{
				Name: pointer.String("service-id"),
				Values: []string{
					serviceID,
				},
			},
		},
	}, func(opts *ec2.Options) {
		opts.Region = r.Network.Spec.Region
	})
	if err != nil {
		return nil, err
	}

	for _, c := range res.VpcEndpointConnections {
		connectionsMap[*c.VpcEndpointId] = newAWSEC2State(c.VpcEndpointState)
	}

	return connectionsMap, nil
}

func (r *reconcileContext) deletePrivateLinkService(ctx context.Context) error {
	if r.Target.Status.SqlEndpoints.PrivateLinkService == nil {
		return nil
	}

	if err := r.refreshTidbClusterStatus(); err != nil {
		return err
	}

	if r.Target.Status.SqlEndpoints.PrivateLinkService.Phase != cloudv1.PrivateLinkServicePhaseDeleting {
		err := r.ensureStatus(func() error {
			r.Target.Status.SqlEndpoints.PrivateLinkService.Phase = cloudv1.PrivateLinkServicePhaseDeleting
			return nil
		})
		if err != nil {
			return err
		}
	}

	updateStatusToNil := func() error {
		return r.ensureStatus(func() error {
			r.Target.Status.SqlEndpoints.PrivateLinkService = nil
			return nil
		})
	}

	updateStatusWithMessage := func(message string) error {
		return r.ensureStatus(func() error {
			r.Target.Status.SqlEndpoints.PrivateLinkService.Message = message
			return nil
		})
	}

	vesc := &ec2v1alpha1.VPCEndpointServiceConfiguration{
		ObjectMeta: metav1.ObjectMeta{
			Name: getPrivateLinkServiceName(r.tc.Namespace, r.tc.Name),
		},
	}
	if err := r.Client.Get(ctx, client.ObjectKeyFromObject(vesc), vesc); err != nil {
		if !apierrors.IsNotFound(err) {
			return err
		}
		// ServiceConfiguration has been deleted
		// Update status
		return updateStatusToNil()
	}

	// Reject all connections before deleting ServiceConfiguration
	if vesc.Status.AtProvider.ServiceConfiguration != nil &&
		vesc.Status.AtProvider.ServiceConfiguration.ServiceID != nil {
		serviceID := *vesc.Status.AtProvider.ServiceConfiguration.ServiceID

		if err := updateStatusWithMessage("reject all connections"); err != nil {
			return err
		}

		connectionsMap, err := r.getAWSConnectionsMap(ctx, serviceID)
		if err != nil {
			return err
		}
		ids := []string{}
		for c, state := range connectionsMap {
			if state != awsEC2StateRejected &&
				state != awsEC2StateDeleted {
				ids = append(ids, c)
			}
		}
		if len(ids) != 0 {
			rejectRes, err := r.AWS.Ec2().RejectVpcEndpointConnections(ctx, &ec2.RejectVpcEndpointConnectionsInput{
				ServiceId:      &serviceID,
				VpcEndpointIds: ids,
			}, func(opts *ec2.Options) {
				opts.Region = r.Network.Spec.Region
			})
			if err != nil {
				return err
			}
			if len(rejectRes.Unsuccessful) != 0 {
				return fmt.Errorf("can't reject all vpc endpoint connections: %v", rejectRes.Unsuccessful)
			}
		}
	}

	// Remove private DNS record
	if err := updateStatusWithMessage("remove private DNS name"); err != nil {
		return err
	}
	rrs := route53v1alpha1.ResourceRecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name: r.privatelinkDnsRecordName(),
		},
	}
	if err := r.Client.Delete(ctx, &rrs); err != nil {
		if !apierrors.IsNotFound(err) {
			return err
		}
	}

	// Remove creation-pending annotation if exist
	// Otherwise it will block deletion of crossplane object
	if vesc.Annotations != nil && vesc.Annotations[AnnotationCrossplaneCreationPending] != "" {
		delete(vesc.Annotations, AnnotationCrossplaneCreationPending)
		if err := r.Client.Update(ctx, vesc); err != nil {
			return err
		}
	}

	// ServiceConfiguration is not deleting, delete it
	if vesc.DeletionTimestamp.IsZero() {
		if err := updateStatusWithMessage("delete crossplane ServiceConfiguration"); err != nil {
			return err
		}

		if err := r.Client.Delete(ctx, vesc); err != nil {
			if !apierrors.IsNotFound(err) {
				return err
			}

			// ServiceConfiguration has been deleted
			// Update status
			return updateStatusToNil()
		}
	}

	return updateStatusToNil()
}