package gke

import (
	"context"
	"fmt"
	"sort"

	gkev1beta1 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	"github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/eks-provider/controllers/tiflow/spec"
	"github.com/tidbcloud/eks-provider/controllers/tiflow/task"
	"github.com/tidbcloud/eks-provider/controllers/tiflow/task/cluster"
	"github.com/tidbcloud/eks-provider/controllers/utils"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	compatcst "github.com/tidbcloud/infra-provider/pkg/compat/consts"
	compatnamer "github.com/tidbcloud/infra-provider/pkg/compat/namer"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
	infrautil "github.com/tidbcloud/infra-provider/pkg/util"
	"github.com/tidbcloud/infra-provider/util/collections"
)

const (
	// this value is meaningless for now, nodes are created when pod triggers node pool scale out
	// the number of nodes will be the same as the number our pods(1 pod on 1 node)
	// keep this if we need to support share node-pool among jobs later.
	// same value as tidb-cluster, for 3 zone, we can have 300 nodes at most.
	maxNodeCountPerZone = 100
)

// NodePool manage lifecycle node pool for gke.
// Note: during test we found the node pool won't scale-in after dataplane-master pods is deleted
// the node pool keeps scale(in or out), but the node count won't become 0. but for tidb node pool
// using nearly same config, it DOES scale-in, not sure why. here is the diff:
// - our node version: 1.23.14-gke.1800, tidb node version: 1.24.9-gke.3200
// - tidb node pool Location policy: Balanced, our node pool don't have it.
// if we want to share node pool by spec code, need to fix this.
type NodePool struct {
	task.BaseTask
	*cluster.BaseContext

	tidbClusterName string
	*v1alpha1.Conditions
}

func NewNodePool(ctx *cluster.BaseContext, tidbClusterName string, objCond *v1alpha1.Conditions) *NodePool {
	return &NodePool{
		BaseContext:     ctx,
		tidbClusterName: tidbClusterName,
		Conditions:      objCond,
	}
}

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

func (t *NodePool) Create(ctx context.Context) deploy.Result {
	if t.Network.Status.Phase == cloudv1.NetworkPhaseHibernating ||
		t.Network.Status.Phase == cloudv1.NetworkPhaseHibernated {
		return deploy.Error(fmt.Errorf("network is in %s phase", t.Network.Status.Phase))
	}
	region := t.Network.Spec.Region
	zones, err := t.Provider.FindRegionZones(region)
	if err != nil {
		return deploy.Error(err)
	}

	networkSpec, err := v1alpha1.GKENetworkSpecForNetwork(t.Network)
	if err != nil {
		return deploy.Error(fmt.Errorf("unable to get network spec"))
	}

	refName := namer.UniqueIdentifier(t.Network.ObjectMeta)
	taints := t.getSpecTaints()
	np := &gkev1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name: t.getNodePoolName(),
		},
	}
	labels := cluster.LabelsForNodeGroup(t.BaseContext, t.tidbClusterName)
	mutateFunc := func() error {
		np.Labels = collections.MergeMaps(np.Labels, labels)
		size, err := util.ParseDiskSizeAsGiB(t.TiflowSpec.SystemVolumeSize)
		if err != nil {
			return err
		}
		sa := compatnamer.LegacyBuildGCPNodePoolServiceAccountEmailName(t.Network, networkSpec.ProjectID)
		np.Spec = gkev1beta1.NodePoolSpec{
			ForProvider: gkev1beta1.NodePoolParameters{
				Autoscaling: &gkev1beta1.NodePoolAutoscaling{
					Enabled:      pointer.Bool(true),
					MaxNodeCount: pointer.Int64(maxNodeCountPerZone),
					MinNodeCount: pointer.Int64(0),
				},
				Config: &gkev1beta1.NodeConfig{
					DiskSizeGb:     pointer.Int64(size),
					DiskType:       pointer.String(utils.GetGcpNodepoolDiskType(t.TiflowSpec.InstanceType.Name)),
					ImageType:      pointer.String("cos_containerd"),
					Labels:         labels,
					MachineType:    &t.TiflowSpec.InstanceType.Name,
					ServiceAccount: &sa,
					OauthScopes: []string{
						"https://www.googleapis.com/auth/cloud-platform",
					},
					Taints: taints,
				},
				Locations: zones,
				Management: &gkev1beta1.NodeManagementSpec{
					// Do it manually
					AutoUpgrade: pointer.Bool(false),
					AutoRepair:  pointer.Bool(false),
				},
				// same as tidb cluster, we don't create nodes on creation of node pool
				// let pod trigger node pool scale out
				InitialNodeCount: pointer.Int64(0),
			},
			ResourceSpec: xpv1.ResourceSpec{ProviderConfigReference: &xpv1.Reference{Name: refName}},
		}
		//In the old path, we used the same crossplane to create the NodePool and GKE clusters,
		//we set the ClusterRef field to let CrossPlane fetch metadata from the manages GKE resources.
		//but in the new path, the NodePool and the GKE are no longer managed through the same CrossPlane,
		//so we need to set the Cluster field to let CrossPlane query the Google API for GKE metadata
		if _, ok := t.Network.Annotations[compatcst.CompatFlagAnnotation]; ok {
			clusterName, err := buildCrossPlaneNodePoolClusterName(t.Network)
			if err != nil {
				return err
			}
			np.Spec.ForProvider.Cluster = clusterName
		} else {
			// use clusterRef in old path
			np.Spec.ForProvider.ClusterRef = &xpv1.Reference{Name: namer.LegacyBuildClusterNameByNetwork(t.Network)}
		}
		return nil
	}

	if _, err2 := controllerutil.CreateOrUpdate(ctx, t.Local, np, mutateFunc); err2 != nil {
		return deploy.Error(err2)
	}
	return deploy.Done()
}

func (t *NodePool) PostCreate(ctx context.Context) deploy.Result {
	if t.Network.Status.Phase == cloudv1.NetworkPhaseHibernating ||
		t.Network.Status.Phase == cloudv1.NetworkPhaseHibernated {
		return deploy.Error(fmt.Errorf("network is in %s phase", t.Network.Status.Phase))
	}
	var g gkev1beta1.NodePool
	nodePoolName := t.getNodePoolName()
	ready := false
	if err := t.Local.Get(ctx, types.NamespacedName{Name: nodePoolName}, &g); err == nil {
		ready = g.Status.AtProvider.Status == gkev1beta1.NodePoolStateRunning
	} else {
		t.Logger.Error(err, "fail to get node pool "+nodePoolName)
	}
	cluster.UpdateCondition(t.Conditions, v1alpha1.TiflowClusterNodeGroupReady, ready)

	if ready {
		return deploy.Done()
	} else {
		return deploy.Error(fmt.Errorf("node pool not ready: %s/%s", nodePoolName, g.Status.AtProvider.Status))
	}
}

func (t *NodePool) Delete(ctx context.Context) deploy.Result {
	selector := cluster.ComponentNodePoolSelector(t.BaseContext)
	if err := t.BaseContext.Local.DeleteAllOf(ctx, &gkev1beta1.NodePool{}, selector); err != nil {
		return deploy.Error(err)
	}
	return deploy.Done()
}

func (t *NodePool) PostDelete(ctx context.Context) deploy.Result {
	ngs := &gkev1beta1.NodePoolList{}
	ops := k8s.ListOpts[*gkev1beta1.NodePool]{
		ObjectList: ngs,
		Unpack:     func() []*gkev1beta1.NodePool { return k8s.MapPtr(ngs.Items) },
	}
	list, err := ops.ListFilterBy(ctx, t.BaseContext.Local,
		func(_ *gkev1beta1.NodePool) bool { return true },
		cluster.ComponentNodePoolSelector(t.BaseContext),
	)
	if client.IgnoreNotFound(err) != nil {
		return deploy.Error(err)
	}
	if len(list) != 0 {
		return deploy.Error(fmt.Errorf("node pool not deleted, remain %d", len(list)))
	}
	return deploy.Done()
}

func (t *NodePool) getNodePoolName() string {
	prefix := spec.TiflowNodegroupPrefix + infrautil.FnvHash(fmt.Sprintf("%s-%s", t.Namer.GetTiflowClusterName(), t.TiflowSpec.SpecCode))
	return namer.UniqueIdentifier(t.Network.ObjectMeta, namer.WithPrefix(prefix), namer.WithLimit(40))
}

func (t *NodePool) getSpecTaints() []*gkev1beta1.NodeTaint {
	var taints []*gkev1beta1.NodeTaint
	for key, value := range t.BaseContext.Namer.Labels() {
		taints = append(taints, &gkev1beta1.NodeTaint{
			Key:    key,
			Value:  value,
			Effect: "NO_SCHEDULE",
		})
	}
	// make taint order stable, seems GKE will update node pool if taint order changed
	// might be caused by gcp cross-plane provider
	sort.Slice(taints, func(i, j int) bool {
		return taints[i].Key < taints[j].Key
	})
	return taints
}

func buildCrossPlaneNodePoolClusterName(network *cloudv1.Network) (string, error) {
	networkspec, err := v1alpha1.GKENetworkSpecForNetwork(network)
	if err != nil {
		return "", err
	}

	return fmt.Sprintf("projects/%s/locations/%s/clusters/%s", networkspec.ProjectID, network.Spec.Region, util.GetAnnotation(network, consts.AnnotationKeyDataplaneName)), nil
}