package aks

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

	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	pingcapErrs "github.com/pingcap/errors"
	"github.com/upbound/provider-azure/apis/containerservice/v1beta1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/eks-provider/api/v1alpha1"
	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/eks-provider/controllers/dm/task"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

// MasterNodePool manage lifecycle of node pool.
type MasterNodePool struct {
	*task.BaseTask
}

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

func NewMasterNodePool(ctx *task.Context) *MasterNodePool {
	return &MasterNodePool{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

func (t *MasterNodePool) Create(ctx context.Context) deploy.Result {
	nodePool, err := t.describeMasterNodePools()
	if err != nil {
		return deploy.Error(err)
	}

	if err := k8s.Apply(ctx, t.Local, nodePool); err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *MasterNodePool) PostCreate(ctx context.Context) deploy.Result {

	if err := isNodePoolReady(ctx, t.BaseTask, t.DMMasterSpec); err != nil {
		return deploy.Error(err).WithRequeuAfter(10 * time.Second)
	}
	t.MasterNodePoolReady = true
	return deploy.Done()
}

func (t *MasterNodePool) Delete(ctx context.Context) deploy.Result {
	nodePool := &v1beta1.KubernetesClusterNodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name: GetAKSMasterNodePoolName(t.GetClusterScopeName(), t.DMJob.Name, t.DMMasterSpec.SpecCode),
		},
	}
	return deploy.PanicOrDone(k8s.DeleteObject(ctx, t.Local, nodePool))
}

func (t *MasterNodePool) PostDelete(ctx context.Context) deploy.Result {
	return deploy.Done()
}

func (t *MasterNodePool) getAKSClusterRef() string {
	return t.EnvProvider.GetClusterName()
}

func (t *MasterNodePool) describeMasterNodePools() (*v1beta1.KubernetesClusterNodePool, error) {
	zoneID, err := t.getZoneID()
	if err != nil {
		return nil, err
	}

	labels := t.GetMasterNodeLabels()
	systemVolumeSize, err := util.ParseDiskSizeAsGiB(t.DMMasterSpec.SystemVolumeSize)
	if err != nil {
		return nil, pingcapErrs.Annotatef(err, "invalid dm master system volume size, specCode: %s, SystemVolumeSize:%s", t.DMMasterSpec.SpecCode, t.DMMasterSpec.SystemVolumeSize)
	}
	nodePool := &v1beta1.KubernetesClusterNodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name:        GetAKSMasterNodePoolName(t.GetClusterScopeName(), t.DMJob.Name, t.DMMasterSpec.SpecCode),
			Labels:      labels,
			Annotations: map[string]string{},
		},
		Spec: v1beta1.KubernetesClusterNodePoolSpec{
			ForProvider: v1beta1.KubernetesClusterNodePoolParameters{
				VMSize:       ptr.To(t.DMMasterSpec.InstanceType.Name),
				OsDiskSizeGb: ptr.To(float64(systemVolumeSize)),
				OsSku:        ptr.To("AzureLinux"),
				Zones:        convertStringSliceToPointerSlice([]string{zoneID}), // https://learn.microsoft.com/en-us/azure/aks/cluster-autoscaler?tabs=azure-cli
				NodeLabels:   convertMapstringToPointerMapString(labels),
				NodeTaints: []*string{
					ptr.To(fmt.Sprintf("dedicated=tidb-%s:NoSchedule", task.ComponentName)),
				},
				EnableAutoScaling: ptr.To(false),
				NodeCount:         ptr.To(float64(t.DMMasterSpec.InstanceNumber)),
				KubernetesClusterID: ptr.To(fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.ContainerService/managedClusters/%s",
					t.Kubernetes.Account,
					t.Kubernetes.Cloud.Azure.ResourceGroup,
					t.getAKSClusterRef(),
				)),
				VnetSubnetID: ptr.To(t.Kubernetes.Cloud.Azure.NodeSubnets[0].Name),
				PodSubnetID:  ptr.To(t.Kubernetes.Cloud.Azure.PodSubnets[0].Name),
				Tags:         t.GetResourceTag(),
				// 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.GetProviderConfigName(),
				},
				ManagementPolicies: []xpv1.ManagementAction{
					xpv1.ManagementActionCreate,
					xpv1.ManagementActionObserve,
					xpv1.ManagementActionUpdate,
					xpv1.ManagementActionDelete,
				},
			},
		},
	}
	return nodePool, nil
}

func (t *MasterNodePool) getZoneID() (string, error) {
	zones := t.Zones()
	if len(zones) < 1 {
		return "", errors.New("zones is required")
	}

	zoneIDs := make([]string, 0, len(zones))
	for _, zone := range zones {
		if pos := strings.Index(zone, "-"); pos != -1 {
			zone = zone[pos+1:]
		}
		if t.Config.DMConfig.Azure.Zone != "" &&
			t.Config.DMConfig.Azure.Zone == zone {
			return zone, nil
		}
		zoneIDs = append(zoneIDs, zone)
	}

	return zoneIDs[0], nil
}

// convertStringSliceToPointerSlice converts a slice of strings to a slice of string pointers.
func convertStringSliceToPointerSlice(input []string) []*string {
	output := make([]*string, len(input))
	for i, v := range input {
		output[i] = ptr.To(v)
	}
	return output
}

// 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 GetAKSMasterNodePoolName(clusterScopeName, dmJobName, specCode string) string {
	return GetAKSNodePoolName(clusterScopeName, dmJobName, specCode, eksv1alpha1.DMMasterMemberType)
}

func GetAKSNodePoolName(clusterScopeName, dmJobName, specCode string, memberType eksv1alpha1.DMMemberType) 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: "nightly-123-123", "1.28"
	originalString := fmt.Sprintf("%s/%s/%s/%s/%s",
		memberType,
		clusterScopeName,
		dmJobName,
		task.ComponentName,
		specCode)

	hash := md5.Sum([]byte(originalString))
	hashString := hex.EncodeToString(hash[:])
	result := fmt.Sprintf("n%s", hashString[:11])
	return result
}

func isNodePoolReady(ctx context.Context, t *task.BaseTask, dmSpec *v1alpha1.DMSpec) error {

	nodePool := &v1beta1.KubernetesClusterNodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name: GetAKSNodePoolName(t.GetClusterScopeName(), t.DMJob.Name, dmSpec.SpecCode, dmSpec.MemberType),
		},
	}
	if err := t.Local.Get(ctx, client.ObjectKeyFromObject(nodePool), nodePool); err != nil {
		t.Logger.Error(err, "failed to get "+string(dmSpec.MemberType)+" node pool cr", "name", nodePool.Name, "specCode", dmSpec.SpecCode)
		return err
	}
	var isReady bool
	for _, condition := range nodePool.Status.Conditions {
		if condition.Type == xpv1.TypeReady && condition.Status == corev1.ConditionTrue {
			isReady = true
		}
	}
	if !isReady {
		t.Logger.Info(string(dmSpec.MemberType)+" nodepool is not ready", "name", nodePool.Name, "specCode", dmSpec.SpecCode)
		return errors.New(string(dmSpec.MemberType) + " nodepool is not ready")
	}

	t.Logger.Info(string(dmSpec.MemberType)+" nodepool is ready", "name", nodePool.Name, "specCode", dmSpec.SpecCode)
	return nil
}

func (t *MasterNodePool) GetResourceTag() map[string]*string {
	resourceTags := t.BaseTask.GetResourceTag()
	resourceTags["cluster_scope"] = ptr.To(t.GetClusterScopeName())
	resourceTags["dmjob_id"] = ptr.To(t.DMJob.Name)
	resourceTags["spec_code"] = ptr.To(t.DMMasterSpec.SpecCode)
	resourceTags["member_type"] = ptr.To(string(t.DMMasterSpec.MemberType))
	return resourceTags
}