package common

import (
	"context"
	"fmt"
	"strings"

	elbv2 "github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2"
	route53v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/route53/v1alpha1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/crossplane/crossplane-runtime/pkg/meta"
	azurenetworkv1beta1 "github.com/upbound/provider-azure/apis/network/v1beta1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	"github.com/tidbcloud/eks-provider/controllers/cluster/task/utils"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/applier"
	"github.com/tidbcloud/infra-provider/pkg/applier/mutatefuncs"
	"github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/controllers/common"
	route53pause "github.com/tidbcloud/infra-provider/pkg/crossplane/route53"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

const (
	maxServiceNameLength = 32

	// AWS
	awsLBNameAnnKey                 = "service.beta.kubernetes.io/aws-load-balancer-name"
	LabelKeyAWSLoadBalancerResource = "service.k8s.aws/resource"
	LabelValAWSLoadBalancerResource = "LoadBalancer"
	LabelKeyAWSLoadBalancerStack    = "service.k8s.aws/stack"
	LabelKeyAWSLoadBalancerCluster  = "elbv2.k8s.aws/cluster"
	// This anno is set to avoid call AWS API multiple times
	// This anno is set in FullService task and maybe used in PrivateLink task
	annoKeyAWSLoadBalancerCanonicalHostedZoneID = "load-balancer.aws.cloud.pingcap.com/canonical-hosted-zone-id"
	annoKeyAWSLoadBalancerARN                   = "load-balancer.aws.cloud.pingcap.com/arn"

	// Azure
	aksPublicLoadBalancerName  = "kubernetes"
	aksPrivateLoadBalancerName = "kubernetes-internal"
)

type TiDBGroupService struct {
	*task.BaseTask
	awsCli            aws.Client
	needsPauseRequeue bool
}

var _ deploy.Task = &TiDBGroupService{}

func NewTiDBGroupService(ctx *task.Context, awsCli aws.Client) *TiDBGroupService {
	// AWSCli is used for AWS service, it can be set to nil if the provider is not AWS
	return &TiDBGroupService{BaseTask: &task.BaseTask{Context: ctx}, awsCli: awsCli}
}

func (t *TiDBGroupService) Name() string {
	return "tidbgroup.TiDBGroupService"
}

func (t *TiDBGroupService) Create(ctx context.Context) deploy.Result {
	t.needsPauseRequeue = false
	remote := t.EnvProvider.GetClient()

	groupNames := []string{}
	fns := []func() (bool, error){}
	for i := range t.Cluster.Spec.ExtraGroups {
		group := &t.Cluster.Spec.ExtraGroups[i]
		groupNames = append(groupNames, group.Name)
		status := &t.Cluster.Status.ExtraEndpoints[i]
		namer := t.TidbGroupNamer(group)

		tlsEnabled, err := t.Context.TidbGroupTLSEnabled(ctx, group)
		if err != nil {
			t.Logger.Error(err, "failed to check if tidb group's tls is enabled")
			return deploy.Error(err)
		}

		fns = append(fns, t.reconcileEndpointFunc(ctx, remote, t.newPrivateEndpoint(group, status, namer, tlsEnabled)))
		fns = append(fns, t.reconcileEndpointFunc(ctx, remote, t.newPublicEndpoint(group, status, namer, tlsEnabled)))
	}

	// clean up lbs with TiDB Group labels, but not in managed in cluster spec
	if err := t.deleteExtraGroups(ctx, groupNames...); err != nil {
		return deploy.Error(err)
	}

	wait, err := combine(fns...)
	if err != nil {
		return deploy.Error(err)
	}
	if wait {
		return deploy.Error(fmt.Errorf("service is not ready")).WithRequeuAfter(defaultWaitDuration)
	}
	if t.needsPauseRequeue {
		return deploy.Done().WithRequeuAfter(route53pause.RequeueAfter)
	}

	return deploy.Done()
}

func (t *TiDBGroupService) Delete(ctx context.Context) deploy.Result {
	// delete all tidb groups by labels
	if err := t.deleteExtraGroups(ctx); err != nil {
		return deploy.Error(err)
	}
	return deploy.Done()
}

func (t *TiDBGroupService) deleteExtraGroups(ctx context.Context, keep ...string) error {
	remote := t.EnvProvider.GetClient()

	// cleanup dns records
	switch t.Config.Provider {
	case cloudv1.ProviderTypeEKS, cloudv1.ProviderTypeGKE:
		dnsList := route53v1alpha1.ResourceRecordSetList{}
		if err := t.Local.List(ctx, &dnsList, client.MatchingLabelsSelector{
			Selector: t.childrenSelector(""),
		}); err != nil {
			return err
		}
		for _, record := range dnsList.Items {
			group, ok := record.GetLabels()[labelKeyExtraGroup]
			if !ok {
				continue
			}

			if groupIsContained(group, keep...) {
				continue
			}

			if err := k8s.DeleteObjectCompletedWithHooks(ctx, t.Local, &record, k8s.HookAutoUnpauseCrossplaneCR); err != nil {
				return err
			}
		}
	case cloudv1.ProviderTypeAKS:
		dnsList := azurenetworkv1beta1.DNSARecordList{}
		if err := t.Local.List(ctx, &dnsList, client.MatchingLabelsSelector{
			Selector: t.childrenSelector(""),
		}); err != nil {
			return err
		}
		for _, record := range dnsList.Items {
			group, ok := record.GetLabels()[labelKeyExtraGroup]
			if !ok {
				continue
			}

			if groupIsContained(group, keep...) {
				continue
			}

			if err := k8s.DeleteObjectCompletedWithHooks(ctx, t.Local, &record, k8s.HookAutoUnpauseCrossplaneCR); err != nil {
				return err
			}
		}
	}

	// cleanup svc
	svcList := corev1.ServiceList{}
	if err := remote.List(ctx, &svcList,
		client.InNamespace(t.Namer.GetTiDBNamespace()),
		client.MatchingLabelsSelector{
			Selector: t.childrenSelector(""),
		},
	); err != nil {
		return err
	}

	needWait := false
	for _, svc := range svcList.Items {
		group, ok := svc.GetLabels()[labelKeyExtraGroup]
		if !ok {
			continue
		}

		if groupIsContained(group, keep...) {
			continue
		}

		if err := k8s.DeleteObjectCompleted(ctx, remote, &svc); err != nil {
			return err
		}

		needWait = true
	}

	if needWait {
		return fmt.Errorf("wait for services deleted")
	}

	return nil
}

func (t *TiDBGroupService) deleteEndpoint(ctx context.Context, remote client.Client, ep *endpoint) error {
	logger := log.FromContext(ctx)
	logger.Info("deleting endpoint", "group", ep.groupName, "dns", ep.dnsName, "svc", ep.svcName)

	switch t.Config.Provider {
	case cloudv1.ProviderTypeEKS:
		return t.deleteAWSEndpoint(ctx, remote, ep)
	case cloudv1.ProviderTypeGKE:
		return t.deleteGKEEndpoint(ctx, remote, ep)
	case cloudv1.ProviderTypeAKS:
		return t.deleteAKSEndpoint(ctx, remote, ep)
	default:
		return fmt.Errorf("unsupported provider: %s", t.Config.Provider)
	}
}

func (t *TiDBGroupService) deleteAWSEndpoint(ctx context.Context, remote client.Client, ep *endpoint) error {
	recordName := ep.dnsName
	if err := k8s.DeleteObjectCompletedWithHooks(ctx, t.Local,
		applier.NewClusterObject[*route53v1alpha1.ResourceRecordSet](recordName),
		k8s.HookAutoUnpauseCrossplaneCR,
	); err != nil {
		return err
	}

	if err := k8s.DeleteObjectCompleted(ctx, remote,
		applier.NewObject[*corev1.Service](t.Namer.GetTiDBNamespace(), ep.svcName),
	); err != nil {
		return err
	}

	return nil
}

func (t *TiDBGroupService) deleteGKEEndpoint(ctx context.Context, remote client.Client, ep *endpoint) error {
	if err := k8s.DeleteObjectCompletedWithHooks(ctx, t.Local,
		applier.NewClusterObject[*route53v1alpha1.ResourceRecordSet](ep.dnsName),
		k8s.HookAutoUnpauseCrossplaneCR,
	); err != nil {
		return err
	}

	if err := k8s.DeleteObjectCompleted(ctx, remote,
		applier.NewObject[*corev1.Service](t.Namer.GetTiDBNamespace(), ep.svcName),
	); err != nil {
		return err
	}

	return nil
}

func (t *TiDBGroupService) deleteAKSEndpoint(ctx context.Context, remote client.Client, ep *endpoint) error {
	recordName := t.getAzureDNSRecordName(ep.dnsName)
	if err := k8s.DeleteObjectCompletedWithHooks(ctx, t.Local,
		applier.NewClusterObject[*azurenetworkv1beta1.DNSARecord](recordName),
		k8s.HookAutoUnpauseCrossplaneCR,
	); err != nil {
		return err
	}

	if err := k8s.DeleteObjectCompleted(ctx, remote,
		applier.NewObject[*corev1.Service](t.Namer.GetTiDBNamespace(), ep.svcName),
	); err != nil {
		return err
	}

	return nil
}

func (t *TiDBGroupService) newPublicEndpoint(
	group *cloudv1.ExtraGroupSpec,
	status *cloudv1.ExtraGroupStatus,
	namer *namer.Namer,
	tlsEnabled bool,
) *endpoint {
	ep := &endpoint{
		isEnabled:       group.TiDB.Replicas != 0 && group.SqlEndpoints.Public.IsEnabled(),
		groupName:       group.Name,
		dnsName:         namer.GetPublicEndpointAddress(),
		dnsDisabled:     group.IsDNSDisabled(),
		svcName:         namer.GetPublicTiDBServiceName(),
		svcNamespace:    namer.GetTiDBNamespace(),
		port:            int(group.SqlEndpoints.GetPublicSQLEndpointPort()),
		targetPort:      group.SqlEndpoints.GetTargetPort(),
		targetComponent: group.SqlEndpoints.GetTargetComponent(),
		// Disable white list of public service
		// We assume that all ips in vpc are allowed to visit public service
		loadBalancerSourceRanges: group.SqlEndpoints.Public.LoadBalancerSourceRanges,
	}

	ep.setStatus = func(set bool) {
		status.Name = group.Name
		if set {
			status.SqlEndpointsStatus.PublicEndpoint = &cloudv1.SqlEndpointStatus{
				Host:       utils.GetEffectiveHostName(ep.dnsName, ep.dnsDisabled),
				Port:       ep.port,
				ResourceID: ep.resourceID,
			}
		} else {
			status.SqlEndpointsStatus.PublicEndpoint = nil
		}
	}

	ep.overlaySvc = func(svc *corev1.Service) {
		svc.Spec.Selector = namer.WithComponent(string(ep.targetComponent)).GetPodLabels()
		svc.Labels = common.ConcatenateTags(svc.Labels, common.ConcatenateTags(cloudv1.TagsFromLabels(t.Cluster.Labels),
			map[string]string{
				common.Cluster:   t.Cluster.Name,
				common.UsedBy:    common.CustomerTiDB,
				common.Component: common.Loadbalance,
			},
		))

		switch t.Config.Provider {
		case cloudv1.ProviderTypeEKS:
			// AWS specific annotations
			// tags use the shared tags with cluster
			tags := cloudv1.TagsFromLabels(t.Cluster.Labels)
			tags = common.ConcatenateTags(tags, map[string]string{
				common.Cluster:   t.Cluster.Name,
				common.Component: common.Loadbalance,
				common.UsedBy:    common.CustomerTiDB,
			})
			tags[consts.TagKeyCrossplaneVendor] = consts.TagValCrossplaneVendor

			annotations := map[string]string{
				"service.beta.kubernetes.io/aws-load-balancer-type":                     "external",
				"service.beta.kubernetes.io/aws-load-balancer-nlb-target-type":          "instance",
				"service.beta.kubernetes.io/aws-load-balancer-scheme":                   "internet-facing",
				"service.beta.kubernetes.io/aws-load-balancer-additional-resource-tags": cloudv1.TagStringForAWS(tags),
				"service.beta.kubernetes.io/aws-load-balancer-target-group-attributes":  "preserve_client_ip.enabled=true",
				"service.beta.kubernetes.io/aws-load-balancer-attributes":               "load_balancing.cross_zone.enabled=true",
			}
			// enable ip target type if dedicated SG feature is enabled
			if t.Cluster.FeatureEnabled(consts.ClusterFeatureDedicatedSG) {
				annotations["service.beta.kubernetes.io/aws-load-balancer-nlb-target-type"] = "ip"
			}

			for k, v := range annotations {
				svc.Annotations[k] = v
			}
			utils.SetNLBAnnoForTiproxy(svc.Annotations, tlsEnabled, group.SqlEndpoints.UseTiProxyAsLBTarget)
		case cloudv1.ProviderTypeGKE:
			annotations := map[string]string{
				// Use external load balancer
				// See https://cloud.google.com/kubernetes-engine/docs/concepts/service-load-balancer-parameters
				"networking.gke.io/load-balancer-type": "External",
			}
			if !t.Cluster.FeatureEnabled(consts.ClusterFeatureRemoveNEGAnnotation) {
				// TODO: Remove this annotation for existing clusters, it will not affect connectivity.
				// To use standalone NEG, we need to enable the gke subsetting, instead of using this annotation.
				// See https://cloud.google.com/kubernetes-engine/docs/how-to/internal-load-balancing#gke-subsetting
				annotations["cloud.google.com/neg"] = fmt.Sprintf(`{"exposed_ports": {"%d":{}}}`, ep.port)
			}

			for k, v := range annotations {
				svc.Annotations[k] = v
			}
		case cloudv1.ProviderTypeAKS:
			// Azure specific annotations
			// do nothing
		}
	}

	return ep
}

func (t *TiDBGroupService) newPrivateEndpoint(
	group *cloudv1.ExtraGroupSpec,
	status *cloudv1.ExtraGroupStatus,
	namer *namer.Namer,
	tlsEnabled bool,
) *endpoint {
	ep := &endpoint{
		isEnabled:                group.TiDB.Replicas != 0 && group.SqlEndpoints.Private != nil && group.SqlEndpoints.Private.IsEnabled(),
		groupName:                group.Name,
		dnsDisabled:              group.IsDNSDisabled(),
		dnsName:                  namer.GetPrivateEndpointAddress(),
		svcName:                  namer.GetPrivateTiDBServiceName(),
		svcNamespace:             namer.GetTiDBNamespace(),
		port:                     int(group.SqlEndpoints.GetPrivateSQLEndpointPort()),
		targetPort:               group.SqlEndpoints.GetTargetPort(),
		targetComponent:          group.SqlEndpoints.GetTargetComponent(),
		loadBalancerSourceRanges: utils.NLBSourceRanges(group.SqlEndpoints.PrivateLinkNLBIPAddressType),
	}

	ep.setStatus = func(enable bool) {
		status.Name = group.Name
		if enable {
			status.SqlEndpointsStatus.PrivateEndpoint = &cloudv1.SqlEndpointStatus{
				Host:       utils.GetEffectiveHostName(ep.dnsName, ep.dnsDisabled),
				Port:       ep.port,
				ResourceID: ep.resourceID,
			}
		} else {
			status.SqlEndpointsStatus.PrivateEndpoint = nil
		}
	}

	ep.overlaySvc = func(svc *corev1.Service) {
		svc.Spec.Selector = namer.WithComponent(string(ep.targetComponent)).GetPodLabels()
		svc.Labels = common.ConcatenateTags(svc.Labels, common.ConcatenateTags(cloudv1.TagsFromLabels(t.Cluster.Labels),
			map[string]string{
				common.Cluster:   t.Cluster.Name,
				common.UsedBy:    common.CustomerTiDB,
				common.Component: common.PrivateLoadbalance,
			},
		))

		switch t.Config.Provider {
		case cloudv1.ProviderTypeEKS:
			// AWS specific annotations
			// tags use the shared tags with cluster
			tags := common.ConcatenateTags(cloudv1.TagsFromLabels(t.Cluster.Labels),
				map[string]string{
					common.Cluster:   t.Cluster.Name,
					common.Component: common.PrivateLoadbalance,
					common.UsedBy:    common.CustomerTiDB,
				})

			// See https://kubernetes-sigs.github.io/aws-load-balancer-controller/v2.4/guide/service/annotations
			annotations := map[string]string{
				"service.beta.kubernetes.io/aws-load-balancer-type":                     "external",
				"service.beta.kubernetes.io/aws-load-balancer-nlb-target-type":          "instance",
				"service.beta.kubernetes.io/aws-load-balancer-scheme":                   "internal",
				"service.beta.kubernetes.io/aws-load-balancer-additional-resource-tags": cloudv1.TagStringForAWS(tags),
				"service.beta.kubernetes.io/aws-load-balancer-target-group-attributes":  "preserve_client_ip.enabled=false",
				"service.beta.kubernetes.io/aws-load-balancer-attributes":               "load_balancing.cross_zone.enabled=true",
			}
			// enable ip target type if dedicated SG feature is enabled
			if t.Cluster.FeatureEnabled(consts.ClusterFeatureDedicatedSG) {
				annotations["service.beta.kubernetes.io/aws-load-balancer-nlb-target-type"] = "ip"
			}
			if group.SqlEndpoints.PrivateLinkNLBIPAddressType != "" {
				annotations[utils.AWSLoadBalancerIPAddressTypeAnnotation] = string(group.SqlEndpoints.PrivateLinkNLBIPAddressType)
			}

			for k, v := range annotations {
				svc.Annotations[k] = v
			}

			utils.SetNLBAnnoToEnablePPV2(t.Cluster, svc.Annotations)
			utils.SetNLBAnnoForTiproxy(svc.Annotations, tlsEnabled, group.SqlEndpoints.UseTiProxyAsLBTarget)

		case cloudv1.ProviderTypeGKE:
			annotations := map[string]string{
				// Use internal load balancer
				// See https://cloud.google.com/kubernetes-engine/docs/concepts/service-load-balancer-parameters
				"networking.gke.io/load-balancer-type": "Internal",
			}
			for k, v := range annotations {
				svc.Annotations[k] = v
			}
		case cloudv1.ProviderTypeAKS:
			annotations := map[string]string{
				// Use internal load balancer
				// See https://learn.microsoft.com/en-us/azure/aks/internal-lb?tabs=set-service-annotations#create-an-internal-load-balancer
				"service.beta.kubernetes.io/azure-load-balancer-internal": "true",
			}

			for k, v := range annotations {
				svc.Annotations[k] = v
			}
		}
	}

	return ep
}

func (t *TiDBGroupService) reconcileEndpointFunc(ctx context.Context, remote client.Client, ep *endpoint) func() (wait bool, _ error) {
	logger := log.FromContext(ctx).WithValues("group", ep.groupName, "dns", ep.dnsName, "svc", ep.svcName)
	return func() (wait bool, _ error) {
		if !ep.isEnabled {
			if err := t.deleteEndpoint(ctx, remote, ep); err != nil {
				return false, err
			}

			ep.setStatus(false)

			return false, nil
		}
		svc := t.NewService(ep)
		clonedSvc := svc.DeepCopy()
		if _, err := controllerutil.CreateOrUpdate(ctx, remote, svc, func() error {
			svc.Spec.LoadBalancerSourceRanges = ep.loadBalancerSourceRanges

			ep.overlaySvc(svc)
			for _, p := range clonedSvc.Spec.Ports {
				mutatefuncs.UpsertServicePort(svc, p)
			}
			return nil
		}); err != nil {
			return false, err
		}

		// update svc status

		if len(svc.Status.LoadBalancer.Ingress) == 0 {
			return true, nil
		}

		address := svc.Status.LoadBalancer.Ingress[0].Hostname
		if address == "" {
			address = svc.Status.LoadBalancer.Ingress[0].IP
		}

		switch t.Config.Provider {
		case cloudv1.ProviderTypeEKS:
			lbName, err := util.GetAWSLBNameFromSVC(svc)
			if err != nil {
				return false, err
			}

			// resource ID
			lbARN, ok := svc.Annotations[annoKeyAWSLoadBalancerARN]
			if !ok {
				arn, err := t.setAWSLoadBalancerARNInAnno(ctx, remote, svc, lbName)
				if err != nil {
					return false, err
				}
				lbARN = arn

			}
			ep.resourceID = lbARN

			// hosted zone ID
			hostedZoneID, ok := svc.Annotations[annoKeyAWSLoadBalancerCanonicalHostedZoneID]
			if !ok {
				id, err := t.setAWSLoadBalancerCanonicalHostedZoneIDInAnno(ctx, remote, svc, lbARN)
				if err != nil {
					return false, err
				}
				hostedZoneID = id
			}

			if ep.dnsDisabled {
				logger.Info("delete dns record due to dnsDisabled is true", "record", ep.dnsName, "rrsName", ep.dnsName)
				if err := k8s.DeleteObjectCompletedWithHooks(ctx, t.Local,
					applier.NewClusterObject[*route53v1alpha1.ResourceRecordSet](ep.dnsName),
					k8s.HookAutoUnpauseCrossplaneCR,
				); err != nil {
					return false, err
				}
			} else {
				// use controllerutil to create or update DNS record, avoid unexpected change for the existed resources
				rrs := t.NewAWSDNSRecord(ep, address, lbName, hostedZoneID)
				if err := t.createRoute53RecordOnceAndPause(ctx, rrs); err != nil {
					return false, err
				}
			}
		case cloudv1.ProviderTypeGKE:
			// DNS
			// use controllerutil to create or update DNS record, avoid unexpected change for the existed resources
			rrs := t.NewGCPDNSRecord(ep, address)
			if err := t.createRoute53RecordOnceAndPause(ctx, rrs); err != nil {
				return false, err
			}

			// resource ID
			frName, err := reconcileFRName(ctx, t.Logger, remote, svc)
			if err != nil {
				return false, err
			}
			ep.resourceID = frName

		case cloudv1.ProviderTypeAKS:
			// DNS
			// use controllerutil to create or update DNS record, avoid unexpected change for the existed resources
			if _, err := controllerutil.CreateOrUpdate(ctx, t.Local, t.NewAzureDNSRecord(ep, address), func() error { return nil }); err != nil {
				return false, err
			}

			// Resource ID
			ep.resourceID = t.getAKSLoadBalancerResourceID(string(svc.GetUID()), ep.svcName == t.Namer.GetPrivateTiDBServiceName())
		}

		ep.setStatus(true)

		return false, nil
	}
}

func (t *TiDBGroupService) createRoute53RecordOnceAndPause(ctx context.Context, rrs *route53v1alpha1.ResourceRecordSet) error {
	if _, err := controllerutil.CreateOrUpdate(ctx, t.Local, rrs, func() error { return nil }); err != nil {
		return err
	}
	needsRequeue, err := route53pause.ReconcilePause(ctx, t.Local, rrs.GetName())
	if err != nil {
		return err
	}
	t.needsPauseRequeue = t.needsPauseRequeue || needsRequeue
	return nil
}

func (t *TiDBGroupService) NewAWSDNSRecord(ep *endpoint, address string, lbName string, elbHostedZoneId string) *route53v1alpha1.ResourceRecordSet {
	// make it competible with the old version in lb management
	rrs := applier.NewClusterObject[*route53v1alpha1.ResourceRecordSet](ep.dnsName)

	// external name will be used as the A record alias address
	rrs.Annotations = map[string]string{
		meta.AnnotationKeyExternalName: ep.dnsName,
	}
	rrs.Labels = t.childrenLabels(ep.groupName)
	// rrs.Annotations[meta.AnnotationKeyExternalName] = dnsName
	rrs.Spec = route53v1alpha1.ResourceRecordSetSpec{
		ResourceSpec: xpv1.ResourceSpec{
			ProviderConfigReference: &xpv1.Reference{
				Name: t.Config.ControlplaneAWSProviderConfig,
			},
		},
		ForProvider: route53v1alpha1.ResourceRecordSetParameters{
			AliasTarget: &route53v1alpha1.AliasTarget{
				DNSName:              fmt.Sprintf("dualstack.%s", address),
				HostedZoneID:         elbHostedZoneId,
				EvaluateTargetHealth: false,
			},
			Type:   "A",
			ZoneID: ptr.To(t.Config.FullManage.PublicHostedZoneID),
		},
	}

	return rrs
}

func (t *TiDBGroupService) NewGCPDNSRecord(ep *endpoint, address string) *route53v1alpha1.ResourceRecordSet {
	rrs := applier.NewClusterObject[*route53v1alpha1.ResourceRecordSet](ep.dnsName)

	rrs.Annotations = map[string]string{
		meta.AnnotationKeyExternalName: ep.dnsName,
	}
	rrs.Labels = t.childrenLabels(ep.groupName)

	rrs.Spec = route53v1alpha1.ResourceRecordSetSpec{
		ResourceSpec: xpv1.ResourceSpec{
			ProviderConfigReference: &xpv1.Reference{
				Name: t.Config.ControlplaneAWSProviderConfig,
			},
		},
		ForProvider: route53v1alpha1.ResourceRecordSetParameters{
			ResourceRecords: []route53v1alpha1.ResourceRecord{
				{
					Value: address,
				},
			},
			Type:   "A",
			TTL:    ptr.To(int64(60)),
			ZoneID: ptr.To(t.Config.FullManage.PublicHostedZoneID),
		},
	}

	return rrs
}

func (t *TiDBGroupService) NewAzureDNSRecord(ep *endpoint, address string) *azurenetworkv1beta1.DNSARecord {
	recordName := ep.dnsName
	rrs := applier.NewClusterObject[*azurenetworkv1beta1.DNSARecord](recordName)

	rrs.Annotations = map[string]string{
		meta.AnnotationKeyExternalName: t.getAzureDNSRecordName(ep.dnsName),
	}
	rrs.Labels = t.childrenLabels(ep.groupName)

	rrs.Spec = azurenetworkv1beta1.DNSARecordSpec{
		ResourceSpec: xpv1.ResourceSpec{
			ProviderConfigReference: &xpv1.Reference{
				Name: t.Config.ControlplaneAzureProviderConfig,
			},
		},
		ForProvider: azurenetworkv1beta1.DNSARecordParameters{
			Records: []*string{
				ptr.To(address),
			},
			ZoneName:          ptr.To(t.Config.FullManage.PublicHostedZoneDomain),
			ResourceGroupName: ptr.To(t.Config.FullManage.AzurePublicHostedResourceGroupName),
			TTL:               ptr.To(float64(60)),
		},
	}

	return rrs
}

func (t *TiDBGroupService) NewService(ep *endpoint) *corev1.Service {
	svc := applier.NewObject[*corev1.Service](ep.svcNamespace, ep.svcName)

	svc.Labels = t.childrenLabels(ep.groupName)
	svc.Annotations = map[string]string{}
	svc.Spec = corev1.ServiceSpec{
		Type:                  corev1.ServiceTypeLoadBalancer,
		SessionAffinity:       corev1.ServiceAffinityNone,
		ExternalTrafficPolicy: corev1.ServiceExternalTrafficPolicyTypeLocal,
		Ports: []corev1.ServicePort{
			{
				Name:       "mysql-client",
				Port:       int32(ep.port),
				TargetPort: ep.targetPort,
				Protocol:   corev1.ProtocolTCP,
			},
		},
		LoadBalancerSourceRanges: ep.loadBalancerSourceRanges,
	}

	return svc
}

func (t *TiDBGroupService) getAKSLoadBalancerResourceID(svcUID string, isPrivateLB bool) string {
	// Azure will add an character 'a' in the head and remove all the '-' to construct load balancer name, add 'a' and then limit the length to 32
	// https://github.com/kubernetes-sigs/cloud-provider-azure/blob/06ec1846c13c3ea6b366a5b888d0fff6a28a77a2/pkg/provider/azure_loadbalancer.go#L4144
	frontendIPConfigurationID := strings.ReplaceAll("a"+svcUID, "-", "")
	if len(frontendIPConfigurationID) > maxServiceNameLength {
		frontendIPConfigurationID = frontendIPConfigurationID[:maxServiceNameLength]
	}
	// Name of the load balancer, when using internal load balancer, the name is kubernetes-internal
	subscriptionID := t.Cluster.Spec.Kubernetes.Account
	resourceGroupName := t.Cluster.Spec.Kubernetes.Cloud.Azure.NodeResourceGroup
	loadBalancerName := aksPublicLoadBalancerName
	if isPrivateLB {
		loadBalancerName = aksPrivateLoadBalancerName
	}

	return fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Network/loadBalancers/%s/frontendIPConfigurations/%s", subscriptionID, resourceGroupName, loadBalancerName, frontendIPConfigurationID)
}

func (t *TiDBGroupService) getAzureDNSRecordName(dnsName string) string {
	return strings.TrimSuffix(dnsName, fmt.Sprintf(".%s", t.Config.FullManage.PublicHostedZoneDomain))
}

func (t *TiDBGroupService) setAWSLoadBalancerARNInAnno(ctx context.Context, remote client.Client, svc *corev1.Service, lbName string) (string, error) {
	arn, err := t.getAWSLoadBalancerARN(ctx, lbName)
	if err != nil {
		return "", err
	}

	if err := k8s.PatchObject(ctx, remote, svc, newARNPatch(arn)); err != nil {
		return "", err
	}

	return arn, nil
}

func (t *TiDBGroupService) getAWSLoadBalancerARN(ctx context.Context, lbName string) (string, error) {
	ar := t.GetAWSAssumedRoleForSDK()
	elbCli := t.awsCli.ELBV2(ar)
	result, err := elbCli.DescribeLoadBalancers(ctx, &elbv2.DescribeLoadBalancersInput{
		Names: []string{lbName},
	})
	if err != nil {
		return "", err
	}

	if len(result.LoadBalancers) == 0 || result.LoadBalancers[0].LoadBalancerArn == nil {
		return "", fmt.Errorf("failed get loadbalancer arn")
	}
	arn := result.LoadBalancers[0].LoadBalancerArn
	return *arn, nil
}

func (t *TiDBGroupService) setAWSLoadBalancerCanonicalHostedZoneIDInAnno(ctx context.Context, remote client.Client, svc *corev1.Service, arn string) (string, error) {
	hostedZoneID, err := t.getLoadBalancerCanonicalHostedZoneID(ctx, arn)
	if err != nil {
		return "", err
	}

	if err := k8s.PatchObject(ctx, remote, svc, newHostedZoneIDPatch(hostedZoneID)); err != nil {
		return "", err
	}

	return hostedZoneID, nil
}

func (t *TiDBGroupService) getLoadBalancerCanonicalHostedZoneID(ctx context.Context, arn string) (string, error) {
	c := t.awsCli.ELBV2(t.GetAWSAssumedRoleForSDK())
	result, err := c.DescribeLoadBalancers(ctx, &elbv2.DescribeLoadBalancersInput{
		LoadBalancerArns: []string{arn},
	})
	if err != nil {
		return "", err
	}
	if len(result.LoadBalancers) == 0 || result.LoadBalancers[0].CanonicalHostedZoneId == nil {
		return "", fmt.Errorf("NLB not found for arn %s", arn)
	}
	return *result.LoadBalancers[0].CanonicalHostedZoneId, nil
}

func (t *TiDBGroupService) childrenLabels(groupName string) map[string]string {
	return childrenLabels(t.Cluster, labelValOwnedByService, groupName)
}

func (t *TiDBGroupService) childrenSelector(groupName string) labels.Selector {
	return childrenSelector(t.Cluster, labelValOwnedByService, groupName)
}

func newARNPatch(arn string) k8s.ObjectPatch {
	return newAnnoPatch(annoKeyAWSLoadBalancerARN, arn)
}

func newHostedZoneIDPatch(hostedZoneID string) k8s.ObjectPatch {
	return newAnnoPatch(annoKeyAWSLoadBalancerCanonicalHostedZoneID, hostedZoneID)
}