package aks

import (
	"context"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"strings"

	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/upbound/provider-azure/apis/containerservice/v1beta1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	commontask "github.com/tidbcloud/eks-provider/controllers/cluster/task/common"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/controllers/common"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
)

const (
	maxNodeCountPerZone int64 = 100
)

type NodePool struct {
	*commontask.NodePool
}

func (t *NodePool) Name() string {
	return "NodePool"
}

// TODO(ideascf): merge it into common/nodepool.go
func NewNodePool(ctx *task.Context) *NodePool {
	return &NodePool{
		NodePool: &commontask.NodePool{
			BaseTask: &task.BaseTask{Context: ctx},
		},
	}
}

func (t *NodePool) Create(ctx context.Context) deploy.Result {
	nps, err := t.desiredNodePools()
	if err != nil {
		return deploy.Panic(err)
	}

	for _, np := range nps {
		instanceType := *np.Spec.ForProvider.VMSize
		zones := np.Spec.ForProvider.Zones
		k8sVersion := t.GetNodeVersion()
		t.Logger.Info("creating nodepool", "instanceType", instanceType,
			"zones", zones, "k8sVersion", k8sVersion)
		if err != nil {
			return deploy.Error(err)
		}

		if _, err = controllerutil.CreateOrUpdate(ctx, t.Local, np, func() error { return nil }); err != nil {
			t.Logger.Error(err, "creating nodepool failed", "instanceType", instanceType,
				"zones", zones, "k8sVersion", k8sVersion)
			return deploy.Error(err)
		}
	}

	return deploy.Done()
}

// desiredNodePools generate nodepools expected to createOrUpdate
func (t *NodePool) desiredNodePools() ([]*v1beta1.KubernetesClusterNodePool, error) {
	zones := t.Zones()
	if len(zones) < 1 {
		return nil, nil
	}
	// zones format is eastus-1, eastus-2, eastus-3
	// crossplane nodepool uses the ordinal as zones parameter
	// So we need to strip the region prefix
	nodepoolZones := make([]string, len(zones))
	for i, z := range zones {
		s := strings.Split(z, "-")
		if len(s) != 2 {
			return nil, nil
		}
		nodepoolZones[i] = s[len(s)-1]
	}

	npsMap := make(map[string]*v1beta1.KubernetesClusterNodePool)
	for _, c := range t.Cluster.GetAllComponents() {
		// TiFlash can be not specified, and the `comp` would be an empty struct
		// Just continue, cleanup will be done separately if it is in the case of scale in to 0
		if c.GetReplicas() < 1 || t.isComponentExpectedToPause(c) {
			continue
		}

		it := c.GetInstanceType()

		for _, zone := range nodepoolZones {
			nodepoolName := t.GetAKSNodePoolName(it, zone)
			npsMap[nodepoolName] = &v1beta1.KubernetesClusterNodePool{
				ObjectMeta: metav1.ObjectMeta{
					Name:        nodepoolName,
					Labels:      convertPtrMap(t.resourceLabels()),
					Annotations: map[string]string{},
				},
				Spec: v1beta1.KubernetesClusterNodePoolSpec{
					ForProvider: v1beta1.KubernetesClusterNodePoolParameters{
						VMSize: &it,
						// Azure uses maxPods to calculate system reserved resources
						// https://learn.microsoft.com/en-us/azure/aks/node-resource-reservations#memory-reservations
						MaxPods:      ptr.To(float64(16)),
						OsDiskSizeGb: ptr.To(float64(100)),
						OsSku:        ptr.To("AzureLinux"),
						Zones:        []*string{ptr.To(zone)}, // https://learn.microsoft.com/en-us/azure/aks/cluster-autoscaler?tabs=azure-cli
						Tags:         t.resourceLabels(),
						NodeLabels:   t.kubernetesLabels(),
						NodeTaints: []*string{
							ptr.To(fmt.Sprintf("dedicated=tidb-%s:NoSchedule", common.ComponentShared)),
						},
						EnableAutoScaling:   ptr.To(true),
						MinCount:            ptr.To(float64(0)),
						MaxCount:            ptr.To(float64(maxNodeCountPerZone)),
						KubernetesClusterID: ptr.To(fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.ContainerService/managedClusters/%s", t.Cluster.Spec.Kubernetes.Account, t.Cluster.Spec.Kubernetes.Cloud.Azure.ResourceGroup, t.GetAKSClusterRef())),
						// https://learn.microsoft.com/en-us/azure/aks/configure-azure-cni-dynamic-ip-allocation
						VnetSubnetID: ptr.To(t.Cluster.Spec.Kubernetes.Cloud.Azure.NodeSubnets[0].Name),
						PodSubnetID:  ptr.To(t.Cluster.Spec.Kubernetes.Cloud.Azure.PodSubnets[0].Name),
						// refer to https://learn.microsoft.com/en-us/azure/aks/use-ultra-disks
						UltraSsdEnabled: ptr.To(false),
						// if it is not set, the default value is 10%, and aks will be frequently update by crossplane
						// because it is not set by crossplane lateinitialize
						// https://github.com/crossplane-contrib/provider-upjet-azure/issues/686
						UpgradeSettings: []v1beta1.KubernetesClusterNodePoolUpgradeSettingsParameters{
							{
								MaxSurge: ptr.To("10%"),
							},
						},
					},
					ResourceSpec: xpv1.ResourceSpec{
						ProviderConfigReference: &xpv1.Reference{
							Name: t.GetAKSProviderConfigName(),
						},
						ManagementPolicies: []xpv1.ManagementAction{
							xpv1.ManagementActionCreate,
							xpv1.ManagementActionObserve,
							xpv1.ManagementActionUpdate,
							xpv1.ManagementActionDelete,
						},
					},
				},
			}
		}
	}

	nps := make([]*v1beta1.KubernetesClusterNodePool, 0)
	for _, np := range npsMap {
		nps = append(nps, np)
	}
	return nps, nil
}

func (t *NodePool) GetAKSClusterRef() string {
	return t.EnvProvider.GetClusterName()
}

func (t *NodePool) isComponentExpectedToPause(c cloudv1.ClusterComponent) bool {
	return c.NeedSuspend(t.Cluster.Spec.Cluster.SuspendAction)
}

// resourceLabels return the labels apply to aks nodes
// The key must only contain lowercase letters ([a-z]), numeric characters ([0-9]), underscores (_) and dashes (-), and must start with a letter.
func (t *NodePool) resourceLabels() map[string]*string {
	l := map[string]string{}

	// Append billing labels
	l[cloudv1.TagLabelPrefix+common.UsedBy] = common.CustomerTiDB
	l[cloudv1.TagLabelPrefix+common.Cluster] = common.ClustersShared
	l[cloudv1.TagLabelPrefix+common.ServiceType] = common.Dedicated
	l[cloudv1.TagLabelPrefix+common.Project] = t.Cluster.GetProjectID()
	l[cloudv1.TagLabelPrefix+common.Tenant] = t.Cluster.GetTenantID()
	if env, ok := t.Cluster.Labels[cloudv1.TagLabelPrefix+common.Environment]; ok {
		l[cloudv1.TagLabelPrefix+common.Environment] = env
	}
	l[consts.K8sVersionLabelKey] = t.GetNodeVersion()

	// Strip from tag prefix
	return convertMapstringToPointerMapString(cloudv1.TagsFromLabels(l))
}

func (t *NodePool) kubernetesLabels() map[string]*string {
	l := t.resourceLabels()

	// NodePool type, in order to make the node selector of tidb components
	// Or it may be scheduled to admin nodepool when the machine type is the same
	l[consts.LabelKeyNodeGroupType] = ptr.To(consts.LabelValNodeGroupTypeDataPlane)

	// Add k8s version label
	l[consts.K8sVersionLabelKey] = ptr.To(t.GetNodeVersion())

	return l
}

// convertMapstringToPointerMapString converts a map[string]string to a map[string]*string
func convertMapstringToPointerMapString(input map[string]string) map[string]*string {
	result := make(map[string]*string)
	for key, value := range input {
		result[key] = ptr.To(value)
	}
	return result
}

func (t *NodePool) GetNodeVersion() string {
	return t.Cluster.Spec.Kubernetes.Version
}

func (t *NodePool) GetAKSNodePoolName(machineType string, zone string) string {
	// https://learn.microsoft.com/en-us/troubleshoot/azure/azure-kubernetes/create-upgrade-delete/aks-common-issues-faq#what-naming-restrictions-are-enforced-for-aks-resources-and-parameters-
	// AKS node pool names must be all lowercase. The names must be 1-12 characters in length for Linux node pools and 1-6 characters for Windows node pools. A name must start with a letter, and the only allowed characters are letters and numbers.

	// example: "dedicated-dev-eastus-af0534b1cc-bb7a0a31", "Standard_DS2_v2", "1.28"
	originalString := fmt.Sprintf("%s/%s/%s/%s", t.GetAKSClusterRef(), machineType, t.GetNodeVersion(), zone)
	hash := md5.Sum([]byte(originalString))
	hashString := hex.EncodeToString(hash[:])
	result := fmt.Sprintf("n%s", hashString[:11])
	return result
}

func convertPtrMap(original map[string]*string) map[string]string {
	newMap := make(map[string]string)
	for key, value := range original {
		if value != nil {
			newMap[key] = *value
		} else {
			newMap[key] = ""
		}
	}
	return newMap
}