package aks

import (
	"context"
	"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"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"

	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"
)

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

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

func NewWorkerNodePool(ctx *task.Context) *WorkerNodePool {
	return &WorkerNodePool{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

func (t *WorkerNodePool) Create(ctx context.Context) deploy.Result {
	nodePool, err := t.describeWorkerNodePools()
	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 *WorkerNodePool) PostCreate(ctx context.Context) deploy.Result {

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

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

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

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

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

	labels := t.GetWorkerNodeLabels()
	systemVolumeSize, err := util.ParseDiskSizeAsGiB(t.DMWorkerSpec.SystemVolumeSize)
	if err != nil {
		return nil, pingcapErrs.Annotatef(err, "invalid dm worker system volume size, specCode: %s, SystemVolumeSize:%s", t.DMWorkerSpec.SpecCode, t.DMWorkerSpec.SystemVolumeSize)
	}
	nodePool := &v1beta1.KubernetesClusterNodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name:        GetAKSWorkerNodePoolName(t.GetClusterScopeName(), t.DMJob.Name, t.DMWorkerSpec.SpecCode),
			Labels:      labels,
			Annotations: map[string]string{},
		},
		Spec: v1beta1.KubernetesClusterNodePoolSpec{
			ForProvider: v1beta1.KubernetesClusterNodePoolParameters{
				VMSize:       ptr.To(t.DMWorkerSpec.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.DMWorkerSpec.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 *WorkerNodePool) 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
}

func GetAKSWorkerNodePoolName(clusterScopeName, dmJobName, specCode string) string {
	return GetAKSNodePoolName(clusterScopeName, dmJobName, specCode, eksv1alpha1.DMWorkerMemberType)
}

func (t *WorkerNodePool) 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.DMWorkerSpec.SpecCode)
	resourceTags["member_type"] = ptr.To(string(t.DMWorkerSpec.MemberType))
	return resourceTags
}