package ack

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

	ackv1alpha1 "github.com/crossplane-contrib/provider-upjet-alibabacloud/apis/ack/v1alpha1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	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/infra-provider/regional-server/pkg/common/constants"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/nodepool/task"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/deploy"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/k8s"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/labels"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/namer"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/nodepool"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
)

const (
	defaultWorkerNodeMaxSize = 100
	defaultWorkerNodeMinSize = 0

	// ackRefLabelKey is a label used for grouping resources belong to a specific ack cluster
	ackRefLabelKey = "ack-reference"
)

type nodePool struct {
	*task.BaseTask
}

func NewNodePool(ctx *task.Context) deploy.Task {
	return &nodePool{BaseTask: &task.BaseTask{Context: ctx}}
}

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

func (t *nodePool) Create(ctx context.Context) deploy.Result {
	t.BeginAt = ptr.To(time.Now())

	if t.APPBox.Status.KubernetesClusterName == "" {
		return deploy.Panic(fmt.Errorf("kubernetes cluster name is empty"))
	}
	if t.APPBox.Status.IAM.NodeRoleName == "" {
		return deploy.Panic(fmt.Errorf("node role name is empty"))
	}

	vswitches := make([]string, 0, len(t.VPC.Status.NodeSubnets))
	for _, subnet := range t.VPC.Status.NodeSubnets {
		if subnet.SubnetID != "" {
			vswitches = append(vswitches, subnet.SubnetID)
		}
	}
	if len(vswitches) == 0 {
		return deploy.Error(fmt.Errorf("node subnets are not ready"))
	}

	metaLabels := util.MergeMaps(t.Namer.Labels(), t.APPBox.Labels)
	metaLabels = util.MergeMaps(metaLabels, labels.NewOwnerLabels(t.NodePool))
	// For compatibility only. It seems not used now
	metaLabels[ackRefLabelKey] = t.APPBox.Status.KubernetesClusterName

	specLabels := map[string]string{
		consts.LabelKeyNodeSpec:   nodepool.NodeSpecLabel(t.NodePool.Spec.CpuCore, t.NodePool.Spec.MemoryGi),
		consts.K8sVersionLabelKey: t.NodePool.Spec.KubernetesVersion,
	}
	maps.Copy(specLabels, t.NodePool.Spec.Labels)

	taints := make([]ackv1alpha1.KubernetesNodePoolTaintsParameters, 0, len(t.NodePool.Spec.Taints)+2)
	userTaints := task.EnsureNodeTypeLabel(t.NodePool.Spec.Taints, task.NormalizeNodeType(t.NodePool.Spec.NodeType))
	if t.NodePool.Spec.WithLocalDisk {
		userTaints = task.EnsureLocalDiskTaint(userTaints)
		task.EnsureLocalDiskLabelsString(specLabels, metaLabels, t.NodePool.Spec.LocalDiskSizeGi)
		specLabels[consts.LabelKeyFastDiskNode] = "pv-raid"
	}
	for _, taint := range userTaints {
		taints = append(taints, ackv1alpha1.KubernetesNodePoolTaintsParameters{
			Effect: ptr.To(task.NormalizeTaintEffect(taint.Effect)),
			Key:    ptr.To(taint.Key),
			Value:  ptr.To(taint.Value),
		})
	}

	userdata, err := t.Config.LoadNodeUserData()
	if err != nil {
		return deploy.Error(err)
	}
	localDiskUserdata, err := t.Config.LoadLocalDiskNodeUserData()
	if err != nil {
		return deploy.Error(err)
	}

	systemDiskSize := t.workerSystemVolumeSize()
	selectedUserdata := userdata
	if t.NodePool.Spec.WithLocalDisk {
		selectedUserdata = localDiskUserdata
	}

	npName := t.k8sNodePoolName()
	np := &ackv1alpha1.KubernetesNodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name:   npName,
			Labels: metaLabels,
		},
		Spec: ackv1alpha1.KubernetesNodePoolSpec{
			ForProvider: ackv1alpha1.KubernetesNodePoolParameters{
				NodePoolName:  ptr.To(npName),
				ClusterIDRefs: &xpv1.Reference{Name: t.APPBox.Status.KubernetesClusterName},
				InstanceTypes: []*string{ptr.To(t.NodePool.Spec.InstanceType)},
				RAMRoleName:   ptr.To(t.APPBox.Status.IAM.NodeRoleName),
				VswitchIds:    util.ToSlicePointer(vswitches),
				ScalingConfig: []ackv1alpha1.ScalingConfigParameters{
					{
						MaxSize: ptr.To(float64(defaultWorkerNodeMaxSize)),
						MinSize: ptr.To(float64(defaultWorkerNodeMinSize)),
					},
				},
				SystemDiskCategory: ptr.To("cloud_essd"),
				SystemDiskSize:     ptr.To(float64(systemDiskSize)),
				Tags:               util.ToMapPointer(task.CloudTags(consts.ProviderTypeACK, t.NodePool, t.DataPlane, t.Config)),
				Taints:             taints,
				Labels:             toACKLabels(specLabels),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: t.ProviderConfigName()},
				ManagementPolicies:      controllers.GetResourceManagementPolicies(),
			},
		},
	}

	if selectedUserdata != "" {
		np.Spec.ForProvider.UserData = ptr.To(selectedUserdata)
	}

	np.Spec.ForProvider.Tags[constants.TagNodeGroupSpecCPU] = ptr.To(strconv.Itoa(int(t.NodePool.Spec.CpuCore)))
	np.Spec.ForProvider.Tags[constants.TagNodeGroupSpecMemoryGi] = ptr.To(strconv.Itoa(int(t.NodePool.Spec.MemoryGi)))
	availableCPU := int(t.NodePool.Spec.AvailableCpuMilliCore / 1000)
	if availableCPU <= 0 {
		availableCPU = int(t.NodePool.Spec.CpuCore)
	}
	np.Spec.ForProvider.Tags[constants.TagNodeGroupAvailableCPU] = ptr.To(strconv.Itoa(availableCPU))
	availableMemoryMi := uint64(t.NodePool.Spec.AvailableMemoryMi)
	if availableMemoryMi == 0 {
		availableMemoryMi = nodepool.ComputeAvailableMemory(uint64(t.NodePool.Spec.MemoryGi) * 1024)
	}
	np.Spec.ForProvider.Tags[constants.TagNodeGroupAvailableMemoryMi] = ptr.To(strconv.FormatUint(availableMemoryMi, 10))

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

	t.NodePool.Status.KubernetesName = t.APPBox.Status.KubernetesClusterName
	t.NodePool.Status.NodeRole = t.APPBox.Status.IAM.NodeRoleName
	t.NodePool.Status.Zones = t.GetNodeZones()

	return deploy.Done()
}

func (t *nodePool) PostCreate(ctx context.Context) deploy.Result {
	np := &ackv1alpha1.KubernetesNodePool{}
	if err := t.Local.Get(ctx, client.ObjectKey{Name: t.k8sNodePoolName()}, np); err != nil {
		return deploy.Error(err)
	}

	condition := np.Status.GetCondition(xpv1.TypeReady)
	if condition.Type == xpv1.TypeReady && condition.Status != corev1.ConditionTrue {
		t.NodePool.Status.Conditions = []metav1.Condition{task.ToMetaCondition(condition)}
		return deploy.Error(fmt.Errorf("waiting for nodepool %s to be ready", np.GetName())).WithRequeuAfter(deploy.DefaultRequeueAfter)
	}
	if condition.Type == xpv1.TypeReady {
		t.NodePool.Status.Conditions = []metav1.Condition{task.ToMetaCondition(condition)}
	}

	if np.Status.AtProvider.DesiredSize != nil {
		if size, err := strconv.Atoi(*np.Status.AtProvider.DesiredSize); err == nil {
			t.NodePool.Status.CurrentReplicas = uint32(size)
		}
	}

	return deploy.Done()
}

func (t *nodePool) Delete(ctx context.Context) deploy.Result {
	nps := new(ackv1alpha1.KubernetesNodePoolList)
	selector := client.MatchingLabels(labels.NewOwnerLabels(t.NodePool))
	if err := t.Local.List(ctx, nps, selector); err != nil {
		return deploy.Error(err)
	}
	for i := range nps.Items {
		np := &nps.Items[i]
		if err := controllers.DeleteUpboundResource(ctx, t.Local, np); err != nil {
			return deploy.Panic(err)
		}
	}
	return deploy.Done()
}

func (t *nodePool) PostDelete(ctx context.Context) deploy.Result {
	nps := new(ackv1alpha1.KubernetesNodePoolList)
	selector := client.MatchingLabels(labels.NewOwnerLabels(t.NodePool))
	if err := t.Local.List(ctx, nps, selector); err != nil {
		return deploy.Panic(err)
	}
	if len(nps.Items) > 0 {
		return deploy.Panic(fmt.Errorf("kubernetesnodepools not deleted: %d remaining", len(nps.Items)))
	}
	return deploy.Done()
}

func (t *nodePool) k8sNodePoolName() string {
	namespacedName := t.APPBox.Namespace + t.APPBox.Name
	resourceNamePrefix := t.Config.ResourceNamePrefix
	if t.NodePool.Spec.ResourceNamePrefix != "" {
		resourceNamePrefix = t.NodePool.Spec.ResourceNamePrefix
	}
	prefix := resourceNamePrefix + "-tidb-worker-" + strings.ReplaceAll(t.NodePool.Spec.InstanceType, ".", "-")
	if t.NodePool.Annotations[consts.AnnotationKeyBuiltinNodePool] != "true" {
		prefix = prefix + "-" + fmt.Sprintf("%dc%dg", t.NodePool.Spec.CpuCore, t.NodePool.Spec.MemoryGi)
	}
	return namer.NewName(namespacedName, prefix, "", -1)
}

func (t *nodePool) workerSystemVolumeSize() int64 {
	return t.Config.APPBoxConfig.WorkerNodeConfig.GetSystemVolumeSize()
}

func toACKLabels(labels map[string]string) []ackv1alpha1.LabelsParameters {
	keys := make([]string, 0, len(labels))
	for key := range labels {
		keys = append(keys, key)
	}
	// The provider models labels as an ordered list, so stable ordering avoids perpetual diffs.
	sort.Strings(keys)

	ret := make([]ackv1alpha1.LabelsParameters, 0, len(labels))
	for _, key := range keys {
		value := labels[key]
		ret = append(ret, ackv1alpha1.LabelsParameters{
			Key:   &key,
			Value: &value,
		})
	}
	return ret
}