package namer

import (
	"crypto/md5"
	"fmt"
	"strings"

	"github.com/pingcap/tidb-operator/pkg/apis/label"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/util/validation"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/util"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
)

const (
	ComponentSelectorKey = "tidb-component"
	NamespaceSelectorKey = "tidb-namespace"
	AppBoxSelectorKey    = "tidbcloud-appbox"

	// ClusterSelectorKey is neither the cluster.Name nor the tc.name.
	// It's mainly used for selecting proper nodes for a TiDBCluster.
	// (even though there are multiple TCs in a cluster, for example multiple tidbgroups in a cluster).
	//
	// For the default TC, it's the cluster.Name.
	// For the TCs for tidbgroups, it's the cluster.Name + "-" + tidbgroups.name.
	ClusterSelectorKey                = "tidbcloud-cluster"
	LegacyNetworkSelectorKey          = "tidbcloud-network"
	LegacyNetworkNamespaceSelectorKey = "tidbcloud-namespace"
	nodeGroupSetOwnerKey              = "tidbcloud-ngset-owner"
	DMMemberSpecLabelKey              = "tidbcloud-dm-member-spec"
	DMMasterLabelKey                  = "tidbcloud-dm-master"
	DMWorkerLabelKey                  = "tidbcloud-dm-worker"
	DMJobNameLabelKey                 = "tidbcloud-dm-job-name"
	ImportJobNameLabelKey             = "tidbcloud-import-job-name"

	maxLimit = 63
	hashSize = 8

	// refer to https://pauldally.medium.com/why-you-try-to-keep-your-deployment-names-to-47-characters-or-less-1f93a848d34c
	DeploymentNameLimit = 47

	// cannot create GCP ServiceAccount name length should between 6 and 30
	GKEServiceAccountNameLimit   = 30
	GCPPSCNAMELimit              = 63
	GKENameLimit                 = 40
	GKEK8sIdentLimit             = 20
	gkeHeterogeneousSegmentLimit = (GKENameLimit - hashSize - 2) / 2

	AWSIAMRoleNameLimit   = 63
	AWSIAMPolicyNameLimit = 127

	nameSeperator    = "-"
	nameSeperatorLen = len(nameSeperator)
)

// Namer gives names that are defined by the spec of a resource.
// Ideally all the names should be constructed by calling the methods of Namer in order to keep the naming consistently
type Namer struct {
	networkName       string
	appBoxName        string
	clusterName       string
	tidbNamespace     string
	tidbName          string
	componentName     string
	nodeGroupSetOwner string
	networkNamespace  string
	podName           string

	// Cluster server address
	publicEndpointAddress      string
	privateEndpointAddress     string
	privatelinkEndpointAddress string
	// GCP private service connect domain
	pscEndpointDomain string
	// domainSuffix is in the format of `{identity}.{hostedZoneDomain}`
	// TiDB is exposed in 3 ways:
	// Public LB: `tidb.{identity}.{hostedZoneDomain}`
	// Private LB: `private-tidb.{identity}.{hostedZoneDomain}`
	// PrivateLink: `privatelink-tidb.{identity}.{hostedZoneDomain}`
	// PrivateServiceConnect is special as endpointName is defined by customer: `{endpointName}.{identity}.{hostedZoneDomain}`
	domainSuffix string
}

func NewNamer() *Namer {
	return &Namer{
		// Consistent with infra-api
		tidbName: consts.ClusterPrefix,
	}
}

// Labels return a map of labels that is generally used for label ASGs/NodeGroups
// or as a selector for selecting ASGs/NodeGroups for a component in a TiDBCluster.
//
// However, the map may also be used as AWS cloud reources `tags` in
// ASG.spec.forProvider.launchTemplateData.tagSpecifications and finally apply to EC2s and k8s nodes.
//
//  1. ClusterSelectorKey + NamespaceSelectorKey can uniquely identify a ASG/NodeGroup or a series of nodes
//     for a TiDBCluster(both the default TC and TCs for tidbgroups).
//  2. There is no k8s version info in this map.
func (n *Namer) Labels() map[string]string {
	res := map[string]string{
		LegacyNetworkSelectorKey:          n.networkName,
		LegacyNetworkNamespaceSelectorKey: n.networkNamespace,
		AppBoxSelectorKey:                 n.appBoxName,
		ClusterSelectorKey:                n.clusterName,
		ComponentSelectorKey:              n.componentName,
		NamespaceSelectorKey:              n.tidbNamespace,
		nodeGroupSetOwnerKey:              n.nodeGroupSetOwner,
		// Add this label to k8s nodes to be compatible with the user alert rules.
		consts.GardenerWorkerPrefixLabelName: NodeGroupNamePrefix(n.clusterName, n.componentName),
	}
	for key, value := range res {
		// when clusterName and componentName are not set, the value of GardenerWorkerPrefixLabelName will be `-`,
		// which is invalid for label value.
		if value == "" || value == "-" {
			delete(res, key)
		}
	}
	return res
}

func (n *Namer) LegacyNetworkLabels() map[string]string {
	res := map[string]string{
		LegacyNetworkSelectorKey:          n.networkName,
		LegacyNetworkNamespaceSelectorKey: n.networkNamespace,
	}
	for key, value := range res {
		if value == "" || value == "-" {
			delete(res, key)
		}
	}
	return res
}

func (n *Namer) GetEKSAutoScalingGroupLabels() map[string]string {
	res := map[string]string{
		ClusterSelectorKey:   n.clusterName,
		ComponentSelectorKey: n.componentName,
		NamespaceSelectorKey: n.tidbNamespace,
	}
	for key, value := range res {
		// when clusterName and componentName are not set, the value of GardenerWorkerPrefixLabelName will be `-`,
		// which is invalid for label value.
		if value == "" || value == "-" {
			delete(res, key)
		}
	}
	return res
}

// Hard-coded by tidb-operator. Do not change.
func (n *Namer) GetPodLabels() map[string]string {
	return map[string]string{
		label.InstanceLabelKey:  n.tidbName,
		label.ComponentLabelKey: n.componentName,
	}
}

func (n *Namer) GetPVCLabels() map[string]string {
	res := map[string]string{
		label.InstanceLabelKey:  n.tidbName,
		label.ComponentLabelKey: n.componentName,
		label.AnnPodNameKey:     n.podName,
	}
	for key, value := range res {
		if value == "" || value == "-" {
			delete(res, key)
		}
	}
	return res
}

func (n *Namer) WithPodName(podName string) *Namer {
	res := *n
	res.podName = podName
	return &res
}

// TODO(ideascf): make this as a private function
// Deprecated: use ForTiDBGroupName instead
func (n *Namer) WithTiDBGroupName(groupName string) *Namer {
	res := *n
	if groupName == "" {
		res.tidbName = consts.ClusterPrefix
		return &res
	}

	res.tidbName = NewName(groupName, "db", "", -1)
	return &res
}

func (n *Namer) WithTiDBNamespace(ns string) *Namer {
	res := *n
	res.tidbNamespace = ns
	return &res
}

func (n *Namer) GetTiDBName() string {
	return n.tidbName
}

func (n *Namer) WithTiDBName(tcName string) *Namer {
	n.tidbName = tcName
	return n
}

func (n *Namer) GetDefaultTiDBName() string {
	return consts.ClusterPrefix
}

func (n *Namer) GetTiDBNamespace() string {
	return n.tidbNamespace
}

func (n *Namer) LegacyWithNetworkName(networkName string) *Namer {
	var res Namer = *n
	res.networkName = networkName
	return &res
}

func (n *Namer) LegacyGetNetworkName() string {
	return n.networkName
}

func (n *Namer) LegacyWithNetworkNamespace(namespace string) *Namer {
	var res Namer = *n
	res.networkNamespace = namespace
	return &res
}

func (n *Namer) LegacyGetNetworkNamespace() string {
	return n.networkNamespace
}

func (n *Namer) WithAPPBoxName(appBoxName string) *Namer {
	var res Namer = *n
	res.appBoxName = appBoxName
	return &res
}

func (n *Namer) WithClusterName(clusterName string) *Namer {
	var res Namer = *n
	res.clusterName = clusterName
	return &res
}

func (n *Namer) GetClusterName() string {
	return n.clusterName
}

func (n *Namer) GetTiDBClusterName() string {
	return n.tidbName
}

func (n *Namer) WithNodeGroupSetOwner(nodeGroupSetName string) *Namer {
	res := *n
	res.nodeGroupSetOwner = nodeGroupSetName
	return &res
}

func (n *Namer) GetNodeGroupSetOwner() string {
	return n.nodeGroupSetOwner
}

// NOTE: This name will be used by internal only, the public service name
// should be generated by GetPublicTiDBServiceName
// TODO: Change the service type from Loadbalancer to ClusterIP
func (n *Namer) GetTiDBServiceName() string {
	// This is hard-coded in tidb-operator/pkg/controller and the operator uses the name everywhere. Not using this name will result in many unexpected failures as seen in backup, restore and import
	return fmt.Sprintf("%s-tidb", n.tidbName)
}

// NOTE: public service SG is used for EKS only
func (n *Namer) GetPublicServiceSGName() string {
	return fmt.Sprintf("%s-%s-sg", n.tidbName, n.clusterName)
}

// NOTE: SGP is used for EKS only
func (n *Namer) GetSGPName() string {
	return fmt.Sprintf("%s-%s-sgp", n.tidbName, n.clusterName)
}

func (n *Namer) GetPublicTiDBServiceName() string {
	return fmt.Sprintf("%s-tidb-public", n.tidbName)
}

func (n *Namer) GetPrivateTiDBServiceName() string {
	return fmt.Sprintf("%s-tidb-private", n.tidbName)
}

func (n *Namer) GetClusterCommonName() string {
	return util.FnvHash(n.GetTiDBNamespace())
}

func (n *Namer) WithComponent(comp string) *Namer {
	res := *n
	res.componentName = comp
	return &res
}

func (n *Namer) DNSNames() []string {
	return []string{
		fmt.Sprintf("%s-%s", n.tidbName, n.componentName),
		fmt.Sprintf("%s-%s.%s", n.tidbName, n.componentName, n.tidbNamespace),
		fmt.Sprintf("%s-%s.%s.svc", n.tidbName, n.componentName, n.tidbNamespace),
		fmt.Sprintf("%s-%s-peer", n.tidbName, n.componentName),
		fmt.Sprintf("%s-%s-peer.%s", n.tidbName, n.componentName, n.tidbNamespace),
		fmt.Sprintf("%s-%s-peer.%s.svc", n.tidbName, n.componentName, n.tidbNamespace),
		fmt.Sprintf("*.%s-%s-peer", n.tidbName, n.componentName),
		fmt.Sprintf("*.%s-%s-peer.%s", n.tidbName, n.componentName, n.tidbNamespace),
		fmt.Sprintf("*.%s-%s-peer.%s.svc", n.tidbName, n.componentName, n.tidbNamespace),
		fmt.Sprintf("*.%s-%s-peer.%s.svc", n.tidbName, n.componentName, n.tidbNamespace),
	}
}

func (n *Namer) GetServiceAccountName() string {
	return fmt.Sprintf("%s-%s", n.tidbName, n.componentName)
}

func (n *Namer) WithClusterServerDNSNames(identity string, publicHostedZoneDomain string) *Namer {
	res := *n
	res.publicEndpointAddress = fmt.Sprintf("%s.%s.%s", "tidb", identity, publicHostedZoneDomain)
	res.privateEndpointAddress = fmt.Sprintf("private-%s", res.publicEndpointAddress)
	res.privatelinkEndpointAddress = fmt.Sprintf("privatelink-%s", res.publicEndpointAddress)
	// <identity>.domain
	res.pscEndpointDomain = fmt.Sprintf("%s.%s", identity,
		publicHostedZoneDomain)
	res.domainSuffix = fmt.Sprintf("%s.%s", identity, publicHostedZoneDomain)

	return &res
}

func (n *Namer) GetPrivateServiceConnectDomain() string {
	return n.pscEndpointDomain
}

// With wildcard DNS name, we can decouple the LB/PrivateLink provision and TLS certification generation
func (n *Namer) GetClusterServerDNSNames(useWildcardDNS bool) []string {
	externalNames := []string{}
	if useWildcardDNS {
		externalNames = []string{fmt.Sprintf("*.%s", n.domainSuffix)}
	} else {
		externalNames = []string{
			n.publicEndpointAddress,
			n.privateEndpointAddress,
			n.privatelinkEndpointAddress,
			// In GCP psc, we can't fully decide the dns name in customer's project
			// The dns is generated with the format: <endpoint_name>.<identity>.domain
			// Use '*' to to adapt to any endpoint name taken by customer
			fmt.Sprintf("*.%s", n.pscEndpointDomain),
		}
	}
	names := []string{
		fmt.Sprintf("%s-%s.%s", n.tidbName, n.componentName, n.tidbNamespace),
		fmt.Sprintf("%s-%s.%s.svc", n.tidbName, n.componentName, n.tidbNamespace),
	}
	names = append(names, externalNames...)
	return names
}

func (n *Namer) GetPublicEndpointAddress() string {
	return n.publicEndpointAddress
}

func (n *Namer) GetPrivateEndpointAddress() string {
	return n.privateEndpointAddress
}

func (n *Namer) GetPrivateLinkEndpointAddress() string {
	return n.privatelinkEndpointAddress
}

// Hard-coded name in the config of TiDB components. We shall not change it.
func (n *Namer) CertSecret() string {
	return fmt.Sprintf("%s-%s-cluster-secret", n.tidbName, n.componentName)
}

// Hard-coded name in the config of TiDB components. We shall not change it.
func (n *Namer) ClientCertSecret() string {
	comp := "cluster"
	if n.componentName != "" {
		comp = n.componentName
	}
	return fmt.Sprintf("%s-%s-client-secret", n.tidbName, comp)
}

func (n *Namer) TidbClientCertSecret() string {
	return fmt.Sprintf("%s-tidb-client-secret", n.tidbName)
}

func (n *Namer) TidbAuthTokenJwksSecret() string {
	return fmt.Sprintf("%s-tidb-auth-token-jwks-secret", n.tidbName)
}

func (n *Namer) TidbAuthTokenJwksSecretInInfraAPI() string {
	return fmt.Sprintf("%s-tidb-auth-token-jwks-secret", n.clusterName)
}

func (n *Namer) BootstrapSQLConfigMapName() string {
	return fmt.Sprintf("%s-bootstrap-sql", n.tidbName)
}

func (n *Namer) BootstrapSQLConfigMapNameInMeta() string {
	return BootstrapSQLConfigMapNameInMeta(n.clusterName)
}

// ForTiDBGroup Create a new namer for a TiDBGroup in the given `cluster`
func (n *Namer) ForTiDBGroup(cluster *cloudv1.Cluster, groupSpec *cloudv1.ExtraGroupSpec, publicHostedZoneDomain string) *Namer {
	return n.
		WithTiDBGroupName(groupSpec.Name).
		WithClusterName(fmt.Sprintf("%s-%s", cluster.Name, groupSpec.Name)).
		WithClusterServerDNSNames(groupSpec.SqlEndpoints.DNS.Identity, publicHostedZoneDomain)
}

func BootstrapSQLConfigMapNameInMeta(clusterID string) string {
	return fmt.Sprintf("%s-bootstrap-sql", clusterID)
}

// ClusterAgentServiceAccountName mirrors the cluster-agent Helm chart's
// default serviceAccountName so IAM and RBAC point at the pod identity.
func ClusterAgentServiceAccountName(clusterName string) string {
	name := "cluster-agent-" + strings.TrimSpace(clusterName)
	if len(name) > maxLimit {
		name = name[:maxLimit]
	}
	return strings.TrimRight(name, "-")
}

type IdentifierOptions func(string) string

func WithPrefix(prefix string) IdentifierOptions {
	return func(name string) string {
		if len(prefix) == 0 || prefix[len(prefix)-1] == '-' {
			return prefix + name
		}
		return prefix + "-" + name
	}
}

// WithLimit limits the generated unique identifier.
// Note this option needs to be the last option for `UniqueIdentifier` function
// namely `WithLimit` should be after `WithPrefix` and other `IdentifierOptions`.
// Otherwise the final generated uid would exceed the limit
func WithLimit(limit int) IdentifierOptions {
	return func(name string) string {
		resultLimit := limit - hashSize - 1
		if len(name) > resultLimit {
			return name[:resultLimit]
		}
		return name
	}
}

func WithHash(size int) IdentifierOptions {
	return func(name string) string {
		hash := fmt.Sprintf("%x", md5.Sum([]byte(name)))
		if size > len(hash) {
			return hash
		}
		if size < 0 {
			size = 0
		}
		return hash[:size]
	}
}

func WithNormalizer() IdentifierOptions {
	return func(name string) string {
		return strings.ReplaceAll(name, "/", "-")
	}
}

func (n *Namer) ServerCertSecret() string {
	return fmt.Sprintf("%s-tidb-server-secret", n.tidbName)
}

// ClusterSelector return a subset of Labels() that can be used as a selector for selecting ASGs/NodeGroups
// for a TiDBCluster(both the default TC and TCs for tidbgroups).
//
// Note: it's **AWS only**.
func (n *Namer) ClusterSelector() client.MatchingLabelsSelector {
	return client.MatchingLabelsSelector{
		Selector: labels.Set{
			NamespaceSelectorKey: n.tidbNamespace,
			ClusterSelectorKey:   n.clusterName,
		}.AsSelector(),
	}
}

func (n *Namer) LegacyNetworkSelector() client.MatchingLabelsSelector {
	return client.MatchingLabelsSelector{
		Selector: labels.Set{
			LegacyNetworkSelectorKey:          n.networkName,
			LegacyNetworkNamespaceSelectorKey: n.networkNamespace,
		}.AsSelector(),
	}
}

func UniqueIdentifierV2(meta metav1.Object, options ...IdentifierOptions) string {
	// See https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#rfc-1035-label-names
	const (
		labelLengthLimit = 63
		resultLimit      = labelLengthLimit - hashSize - 1
	)

	namespacedName := fmt.Sprintf("%s/%s", meta.GetNamespace(), meta.GetName())
	hash := fmt.Sprintf("%x", md5.Sum([]byte(namespacedName)))[:hashSize]

	// The 8-digit hash will not be trimmed. Keep the namespaced name if possible
	result := meta.GetName()
	if meta.GetNamespace() != "" {
		result = fmt.Sprintf("%s-%s", meta.GetNamespace(), meta.GetName())
	}
	for _, opt := range options {
		result = opt(result)
	}
	if len(result) > resultLimit {
		result = result[:resultLimit]
	}
	return fmt.Sprintf("%s-%s", result, hash)
}

func UniqueIdentifier(meta metav1.ObjectMeta, options ...IdentifierOptions) string {
	return UniqueIdentifierV2(&meta, options...)
}

func NodeGroupSetOwnerSelector(ngSetName string) client.MatchingLabelsSelector {
	return client.MatchingLabelsSelector{
		Selector: labels.Set{
			nodeGroupSetOwnerKey: ngSetName,
		}.AsSelector(),
	}
}

// GenExternalID generates AWS externalID from Network CR namespace and name
// It's only used for clusters created before 3-layer arch
func GenExternalID(ns, name string) string {
	h := md5.New()
	h.Write([]byte(ns + "/" + name))
	bs := h.Sum(nil)
	return fmt.Sprintf("%x", bs)
}

/* Naming rules below */

func NodeGroupName(cluster string, component string, zone string) string {
	s := strings.Split(zone, "-")
	zoneSuffix := s[len(s)-1]
	return fmt.Sprintf("%s-%s", NodeGroupNamePrefix(cluster, component), zoneSuffix)
}

func NodeGroupNamePrefix(cluster string, componentType string) string {
	return fmt.Sprintf("%s-%s", cluster, componentType)
}

func NodesSubnetName(vpcName string, zone string) string {
	return vpcName + "-nodes-" + zone
}

func PublicSubnetName(vpcName string, zone string) string {
	return vpcName + "-public-" + zone
}

func PrivateSubnetName(vpcName string, zone string) string {
	return vpcName + "-private-" + zone
}

func IsPrivateSubnet(name string) bool {
	return strings.Contains(name, "-private-")
}

func GCPNodePoolServiceAccountNameOptions() []IdentifierOptions {
	return []IdentifierOptions{
		WithPrefix("access-sa-"),
		WithLimit(GKEServiceAccountNameLimit),
	}
}

func GCPNodePoolServiceAccountKeySecretNameOptions() []IdentifierOptions {
	return []IdentifierOptions{
		WithPrefix("access-sa-key"),
		WithLimit(GKEServiceAccountNameLimit),
	}
}

func LegacyBuildGCPNodePoolServiceAccountNameByNetwork(network *cloudv1.Network) string {
	return UniqueIdentifier(network.ObjectMeta, GCPNodePoolServiceAccountNameOptions()...)
}

func LegacyBuildGCPNodePoolServiceAccountEmailName(network *cloudv1.Network, gcpProjectName string) string {
	return fmt.Sprintf("%s@%s.iam.gserviceaccount.com", LegacyBuildGCPNodePoolServiceAccountNameByNetwork(network), gcpProjectName)
}

func LegacyBuildServiceAccountKeyNameByNetwork(network *cloudv1.Network) string {
	return UniqueIdentifier(network.ObjectMeta, GCPNodePoolServiceAccountKeySecretNameOptions()...)
}

// SharedBucketReadonlyRoleName generates the name of the AWS IAM role for accessing shared bucket in readonly.
// This role is usually used by control-plane, for example, downloading DB audit log files from shared bucket.
func SharedBucketReadonlyRoleName(cluster *cloudv1.Cluster) string {
	return NewName(cluster.ClusterScopeName(), "sb-ro", "", AWSIAMRoleNameLimit)
}

// SharedBucketAuditLogWriterRoleName generates the name of the AWS IAM role for write DB audit log in shared bucket
func SharedBucketAuditLogWriterRoleName(cluster *cloudv1.Cluster) string {
	return NewName(cluster.ClusterScopeName(), "sb-audit-log-writer", "", AWSIAMRoleNameLimit)
}

// SharedBucketReadonlySAName generates the name of the GCP IAM SA for accessing shared bucket in readonly.
// This role is usually used by control-plane, for example, downloading DB audit log files from shared bucket.
// Note: As the nodepool is created per dataplane, so that the SA
func SharedBucketReadonlySAName(cluster *cloudv1.Cluster) string {
	return NewName(cluster.Spec.Kubernetes.Name, "sb-ro", "", GKEServiceAccountNameLimit)
}

func SharedBucketReadonlySAEmail(cluster *cloudv1.Cluster) string {
	return fmt.Sprintf("%s@%s.iam.gserviceaccount.com", SharedBucketReadonlySAName(cluster), cluster.Spec.Kubernetes.Account)
}

// ClusterServerIAMSAEmail generates the name of GCP IAM SA used by the cluster-server.
func ClusterServerIAMSAEmail(saName, controlPlaneProject string) string {
	return fmt.Sprintf("%s@%s.iam.gserviceaccount.com", saName, controlPlaneProject)
}

// BuildPSCNameByNetwork can only be used for gke so11y PSC endpoint name.
func BuildPSCNameByNetwork(network *cloudv1.Network, prefix string) string {
	return UniqueIdentifier(network.ObjectMeta, WithPrefix(prefix), WithLimit(GCPPSCNAMELimit))
}

// BuildClusterNameByNetwork return the name of EKS or GKE cluster
// Deprecated: Network will be removed, this function will not be called in cluster controller again
func LegacyBuildClusterNameByNetwork(network *cloudv1.Network) string {
	if network.Spec.Provider == consts.ProviderTypeGKE {
		return UniqueIdentifier(network.ObjectMeta, WithLimit(GKENameLimit))
	}
	return UniqueIdentifier(network.ObjectMeta)
}

// GKENodePoolNameForCluster returns the current-version GKE TiDB NodePool name
// for the given machine type. Prefer this helper when validating persisted
// intent against a Cluster so callers do not duplicate node-version fallback
// rules.
func GKENodePoolNameForCluster(cluster *cloudv1.Cluster, machineType string) string {
	if cluster == nil {
		return ""
	}
	return GKENodePoolName(cluster, machineType, GKENodeVersionForCluster(cluster))
}

// GKEHeterogeneousNodePoolNameForCluster returns the current-version GKE
// heterogeneous NodePool name for the given machine type. It deliberately uses
// a separate machine-type namespace so ordinary and heterogeneous NodePools can
// coexist for the same machine type within one Kubernetes cluster scope.
func GKEHeterogeneousNodePoolNameForCluster(cluster *cloudv1.Cluster, machineType string) string {
	if cluster == nil {
		return ""
	}
	return GKEHeterogeneousNodePoolName(cluster, machineType, GKENodeVersionForCluster(cluster))
}

// GKENodeVersionForCluster returns the node version used in GKE NodePool names.
// Status is preferred because it reflects the actual cluster version; the
// control-plane-version annotation is the create/import fallback.
func GKENodeVersionForCluster(cluster *cloudv1.Cluster) string {
	if cluster == nil {
		return ""
	}
	status, err := cloudv1.GKEClusterStatusFromCluster(cluster)
	if err == nil && strings.TrimSpace(status.NodeVersion) != "" {
		return strings.TrimSpace(status.NodeVersion)
	}
	return strings.TrimSpace(cluster.GetAnnotations()[consts.AnnoKeyControlPlaneVersion])
}

// GKENodePoolName returns the GKE TiDB NodePool name used by the shared
// nodepool path. Keep this helper lightweight so packages that only need naming
// do not import the full nodepool package and its cloud client dependencies.
func GKENodePoolName(cluster *cloudv1.Cluster, machineType, k8sVersion string) string {
	normalizedVersion := NormalizeGKENodeVersion(k8sVersion)
	if cluster.Spec.NetworkName != "" {
		return UniqueIdentifier(metav1.ObjectMeta{
			Name:      cluster.Spec.NetworkName,
			Namespace: cluster.Namespace,
		}, gkeNodePoolNameOptions(normalizedVersion, machineType)...)
	}

	if cluster.FeatureEnabled(consts.ClusterFeatureGCP3LayerNodePool) {
		return NewName(cluster.Spec.Kubernetes.Name, machineType, normalizedVersion, GKENameLimit)
	}

	return NewName(cluster.ClusterScopeName(), machineType, normalizedVersion, GKENameLimit)
}

func GKEHeterogeneousNodePoolName(cluster *cloudv1.Cluster, machineType, k8sVersion string) string {
	if cluster == nil {
		return ""
	}

	heterogeneousMachineType := "h-" + strings.TrimSpace(machineType)
	normalizedVersion := NormalizeGKENodeVersion(k8sVersion)
	if cluster.Spec.NetworkName != "" {
		return UniqueIdentifier(metav1.ObjectMeta{
			Name:      cluster.Spec.NetworkName,
			Namespace: cluster.Namespace,
		}, gkeNodePoolNameOptions(normalizedVersion, heterogeneousMachineType)...)
	}

	identity := cluster.ClusterScopeName()
	if cluster.FeatureEnabled(consts.ClusterFeatureGCP3LayerNodePool) {
		identity = cluster.Spec.Kubernetes.Name
	}
	return gkeHeterogeneousNodePoolName(identity, heterogeneousMachineType, normalizedVersion)
}

func gkeHeterogeneousNodePoolName(identity, machineType, normalizedVersion string) string {
	candidate := NewName(identity, machineType, normalizedVersion, GKENameLimit)
	if len(candidate) <= GKENameLimit && len(validation.IsDNS1035Label(candidate)) == 0 {
		return candidate
	}

	return joinIgnoreEmpty("-",
		compactGKENameSegment(machineType, gkeHeterogeneousSegmentLimit),
		shortHash(identity),
		compactGKENameSegment(normalizedVersion, gkeHeterogeneousSegmentLimit),
	)
}

func compactGKENameSegment(segment string, limit int) string {
	if limit <= 0 {
		return ""
	}
	original := strings.TrimSpace(segment)
	segment = normalizeGKENameSegment(segment)
	if len(segment) <= limit {
		return segment
	}
	hash := shortHash(original)
	keep := limit - hashSize - 1
	if keep <= 0 {
		return hash[:limit]
	}
	prefix := strings.TrimRight(segment[:keep], "-")
	if prefix == "" {
		return hash[:limit]
	}
	return prefix + "-" + hash
}

func normalizeGKENameSegment(segment string) string {
	segment = strings.ToLower(strings.TrimSpace(segment))
	var builder strings.Builder
	builder.Grow(len(segment))
	for _, c := range segment {
		if (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '-' {
			builder.WriteRune(c)
			continue
		}
		builder.WriteRune('-')
	}
	return strings.Trim(builder.String(), "-")
}

func shortHash(value string) string {
	return fmt.Sprintf("%x", md5.Sum([]byte(value)))[:hashSize]
}

func gkeNodePoolNameOptions(normalizedVersion, machineType string) []IdentifierOptions {
	return []IdentifierOptions{
		WithPrefix(normalizedVersion),
		WithPrefix(machineType),
		WithLimit(GKENameLimit),
	}
}

// NormalizeGKENodeVersion converts a GKE node version to the suffix-safe form
// used in GKE NodePool names.
func NormalizeGKENodeVersion(version string) string {
	var normalized string
	for _, c := range version {
		if c != '.' && c != '-' && c != '_' {
			normalized += string(c)
		}
	}
	return normalized
}

// Deprecated: use BuildOverProvisioningNameV2 instead
// BuildOverProvisioningName returns the name for over-provisioning deployment.
func BuildOverProvisioningName(suffix string) string {
	// Azure instance type name is in the format of "Standard_D2s_v3".
	// Replace "_" with "-" to avoid names like "over-provisioning-tidb-Standard_D2s_v3".
	suffix = strings.ReplaceAll(suffix, "_", "-")
	suffix = strings.ToLower(suffix)
	return fmt.Sprintf("over-provisioning-%s", suffix)
}

func BuildOverProvisioningNameV2(compType cloudv1.ComponentType, groupId, instanceType string) string {
	identity := fmt.Sprintf("%s-%s", groupId, instanceType)
	out := NewNameHashOnlyExceed(identity, "op", string(compType), DeploymentNameLimit)

	return normalizeOPDeploymentName(out)
}

// BuildOverProvisioningNameForStandbyNodeV2 returns the name for over-provisioning deployment used to create standby nodes.
func BuildOverProvisioningNameForStandbyNodeV2(compType cloudv1.ComponentType, groupId, zone string) string {
	identity := fmt.Sprintf("%s-%s", groupId, zone)
	out := NewNameHashOnlyExceed(identity, "op-standby", string(compType), DeploymentNameLimit)

	return normalizeOPDeploymentName(out)
}

func normalizeOPDeploymentName(name string) string {
	// Azure instance type name is in the format of "Standard_D2s_v3".
	// Replace "_" with "-" to avoid names like "over-provisioning-tidb-Standard_D2s_v3".
	name = strings.ToLower(name)
	name = strings.ReplaceAll(name, "_", "-")
	return name
}

// HasOverProvisioningNamePrefixV2 returns true if the name has the prefix of over-provisioning deployment.
func HasOverProvisioningNamePrefixV2(name string) bool {
	return strings.HasPrefix(name, "op-")
}

// Deprecated: use HasOverProvisioningNamePrefixV2 in pair with BuildOverProvisioningNameV2
// HasOverProvisioningNamePrefix returns true if the name has the prefix of over-provisioning deployment.
func HasOverProvisioningNamePrefix(name string) bool {
	return strings.HasPrefix(name, "over-provisioning-")
}

// A name is in format like this: prefix-trim(identity, length)-hash-suffix
// Limit is the length limit of whole name, it must bigger than 10 and less than 63
// 10 is hashSize + len('-') + 1
// 63 is label name limit, see https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#rfc-1035-label-names
//
// NOTE: Please use a meaningful prefix and suffix to generate name
func NewName(identity, prefix, suffix string, limit int) string {
	res := ""

	if limit > maxLimit || limit < 0 {
		limit = maxLimit
	}

	length := limit - hashSize - 1
	if len(prefix) != 0 {
		length = length - len(prefix) - 1
	}
	if len(suffix) != 0 {
		length = length - len(suffix) - 1
	}

	if limit <= 0 {
		panic("limit is too small or prefix and suffix are too long")
	}

	ps := []NameProcessor{
		AddPrefix(prefix),
		AddIdentityWithLimitedLength(length),
		AddHash(hashSize),
		AddSuffix(suffix),
	}
	for _, p := range ps {
		res = p.Process(identity, res)
	}

	return res
}

// NewNameHashOnlyExceed is like NewName, but it will not hash the identity if total length doesn't exceeds the limit
func NewNameHashOnlyExceed(identity, prefix, suffix string, limit int) string {
	res := ""

	if limit > maxLimit {
		limit = maxLimit
	}
	if limit < 0 {
		panic(fmt.Sprintf("invalid limit %d", limit))
	}

	length := limit
	if len(prefix) != 0 {
		length = length - len(prefix) - nameSeperatorLen
	}
	if len(suffix) != 0 {
		length = length - len(suffix) - nameSeperatorLen
	}

	if length <= hashSize {
		panic(fmt.Sprintf("limit is too small or prefix and suffix are too long, remain length for identity is %d", length))
	}

	ps := []NameProcessor{
		HashIfExceed(length),
		AddPrefix(prefix),
		AddSuffix(suffix),
	}
	for _, p := range ps {
		res = p.Process(identity, res)
	}

	return res
}

type NameProcessor interface {
	Process(original, result string) string
}

type NameProcessorFunc func(original, result string) string

func (p NameProcessorFunc) Process(original, result string) string {
	return p(original, result)
}

func HashIfExceed(maxLen int) NameProcessor {
	return NameProcessorFunc(func(original, result string) string {
		if len(original) <= maxLen {
			return joinIgnoreEmpty(nameSeperator, result, original)
		}

		// if original: '01234567890123', maxLen: 12, hashSize: 8
		// then keep: '012', out: '012-<hash>'
		keep := original[:maxLen-hashSize-nameSeperatorLen]
		// hash the full original string to reduce the collision rate
		hash := fmt.Sprintf("%x", md5.Sum([]byte(original)))
		out := keep + nameSeperator + hash[:hashSize]
		return joinIgnoreEmpty(nameSeperator, result, out)
	})
}

func AddHash(size int) NameProcessor {
	return NameProcessorFunc(func(original, result string) string {
		hash := fmt.Sprintf("%x", md5.Sum([]byte(original)))
		if size > len(hash) {
			size = len(hash)
		}
		if size < 0 {
			size = 0
		}
		return joinIgnoreEmpty("-", result, hash[:size])
	})
}

func AddIdentityWithLimitedLength(length int) NameProcessor {
	return NameProcessorFunc(func(original, result string) string {
		if length > len(original) {
			length = len(original)
		}
		if length < 0 {
			length = 0
		}
		return joinIgnoreEmpty("-", result, original[:length])
	})
}

func AddPrefix(prefix string) NameProcessor {
	return NameProcessorFunc(func(_, result string) string {
		return joinIgnoreEmpty("-", prefix, result)
	})
}

func AddSuffix(suffix string) NameProcessor {
	return NameProcessorFunc(func(_, result string) string {
		return joinIgnoreEmpty("-", result, suffix)
	})
}

func joinIgnoreEmpty(sep string, parts ...string) string {
	nparts := []string{}
	for _, part := range parts {
		if len(part) != 0 {
			nparts = append(nparts, part)
		}
	}

	return strings.Join(nparts, sep)
}