package service

import (
	"context"
	"fmt"

	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/samber/lo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	appv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-agent/v1"
	apicommon "github.com/tidbcloud/infra-provider/api/tidbcloud/common"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/pingcap/label"
)

func (s *AgentService) ListNodes(ctx context.Context, req *apiv1.ListNodesReq) (*apiv1.ListNodesResp, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	logger.Info("list nodes")
	nodes, err := s.getNodesByClusterId(ctx, req.ClusterId)
	if err != nil {
		return nil, err
	}

	out := lo.MapToSlice(nodes, func(k string, v *corev1.Node) *apiv1.Node {
		return convertNode(v)
	})
	return &apiv1.ListNodesResp{Nodes: out}, nil
}

func (s *AgentService) GetNode(ctx context.Context, req *apiv1.GetNodeReq) (*apiv1.Node, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId), zap.String("node_name", req.NodeName))
	logger.Info("get node")
	var node corev1.Node
	if err := s.dataplaneCli.Get(ctx, client.ObjectKey{Name: req.NodeName}, &node); err != nil {
		return nil, fmt.Errorf("failed to get node: %w", err)
	}
	return convertNode(&node), nil
}

func (s *AgentService) IsNodeReadyForScaleUpDown(ctx context.Context, req *apiv1.IsNodeReadyForScaleUpDownReq) (*apiv1.IsNodeReadyForScaleUpDownResp, error) {
	tcs, err := s.getTCsByClusterId(ctx, req.ClusterId)
	if err != nil {
		return nil, fmt.Errorf("get tcs: %w", err)
	}

	var out []*apiv1.IsNodeReadyForScaleUpDownResp_ComponentResult
	switch s.provider {
	case "aws":
		out, err = s.calcComponentNodeReadyResultForAWS(ctx, req.ClusterId, tcs, req.ComponentTypes)
	case "gcp":
		out, err = s.calcComponentNodeReadyResultForGCP(ctx, req.ClusterId, tcs, req.ComponentTypes)
	case "kind":
		// kind doesn't support this API, just a dummy implement
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &apiv1.IsNodeReadyForScaleUpDownResp{ComponentResults: out}, nil
}

// calcComponentNodeReadyResultForGCP calculates component node ready result for clusters running GCP.
// Due to we use shared nodepool per instanceType, nodes of different components or clusters may be provisioned by the same nodepool.
// So if you try to match nodes by node-selector of component, nodes owned by other components or clusters may be included.
//
// Fortunately, cluster-controller will create a set of OverProvisioning deployments to ensure cluster-autoscaler scale out expected nodes.
// So we can use the Ready result of OverProvisioning pods as the result of nodes.
//
// An example of `tc.spec.tikv.nodeSelector`
//
//	nodeSelector:
//	  cloud.pingcap.com/node-group-type: data-plane
//	  node.kubernetes.io/instance-type: e2-medium
//	  tidbcloud.com/node-pool-k8s-version: "1.28"
func (s *AgentService) calcComponentNodeReadyResultForGCP(ctx context.Context, clusterId string, tcs map[string]*v1alpha1.TidbCluster, compTypes []apicommon.ComponentType) ([]*apiv1.IsNodeReadyForScaleUpDownResp_ComponentResult, error) {
	// NOTE: For now, we don't support tidb-groups on GCP/GKE.
	logger := tilog.WithContextL(ctx)

	var out []*apiv1.IsNodeReadyForScaleUpDownResp_ComponentResult
	for _, tc := range tcs {
		for _, comp := range tc.AllComponentSpec() {
			compType := comp.MemberType().String()
			if !lo.Contains(compTypes, convertComponentTypeToAPI(compType)) {
				logger.Debug("skip for component", zap.String("comp", compType))
				continue
			}

			var list appv1.DeploymentList
			if err := s.dataplaneCli.List(ctx, &list, client.InNamespace(tc.Namespace), client.MatchingLabels(buildOverProvisioningDeploymentLabels(tc.Name, compType))); err != nil {
				return nil, fmt.Errorf("list over provisioning deployment: %w", err)
			}
			result := &apiv1.IsNodeReadyForScaleUpDownResp_ComponentResult{
				TcName:        tc.Name,
				ComponentType: convertComponentTypeToAPI(compType),
				Ready:         false,
				ReadyReason:   "unknown",
			}
			out = append(out, result)

			if cnt := len(list.Items); cnt != 1 {
				result.ReadyReason = fmt.Sprintf("len(deployments)=%d", cnt)
				logger.Error("over provisioning deployment count is not 1", zap.String("tc", tc.Name), zap.String("comp", compType), zap.Int("cnt", cnt))
				continue
			}
			desiredNum := ptr.Deref(list.Items[0].Spec.Replicas, 0)
			actualNum := list.Items[0].Status.ReadyReplicas
			result.Ready = desiredNum == actualNum
			result.ReadyReason = fmt.Sprintf("desired: %d, actual: %d", desiredNum, actualNum)
		}
	}
	return out, nil
}

// calcComponentNodeReadyResultForAWS calculates component node ready result for clusters running AWS.
// Thanks to that we use self-managed(or managed) nodegroup per cluster, component and instanceType, just using node selector to match
// nodes is enough to check whether nodes of component are ready.
// An example of `tc.spec.tikv.nodeSelector`
//
//	nodeSelector:
//	  gardener.wg.prefix.name: 3842079-tikv
//	  tidb-component: tikv
//	  tidb-namespace: tidb3842079
//	  tidbcloud-cluster: "3842079"
//	  tidbcloud-namespace: nightly-ms-43562
//	  tidbcloud-network: eks-us-west-2
func (s *AgentService) calcComponentNodeReadyResultForAWS(ctx context.Context, clusterId string, tcs map[string]*v1alpha1.TidbCluster, compTypes []apicommon.ComponentType) ([]*apiv1.IsNodeReadyForScaleUpDownResp_ComponentResult, error) {
	componentNodeInfos, err := s.buildComponentNodeInfos(ctx, tcs, compTypes)
	if err != nil {
		return nil, fmt.Errorf("buildComponentNodeInfos: %w", err)
	}
	nodes, err := s.getNodesByClusterId(ctx, clusterId)
	if err != nil {
		return nil, err
	}

	for _, info := range componentNodeInfos {
		for _, no := range nodes {
			if info.nodeSelector.Matches(labels.Set(no.Labels)) &&
				// nodes belong to previous nodegroup have been marked as SchedulingDisabled during scale up/down. filter out them
				no.Spec.Unschedulable == false {
				info.actualNodeNum += 1
			}
		}
	}
	var out []*apiv1.IsNodeReadyForScaleUpDownResp_ComponentResult
	for _, info := range componentNodeInfos {
		out = append(out, &apiv1.IsNodeReadyForScaleUpDownResp_ComponentResult{
			TcName:        info.tcName,
			ComponentType: convertComponentTypeToAPI(info.componentType),
			Ready:         info.actualNodeNum >= info.desiredNodeNum,
			ReadyReason:   fmt.Sprintf("desired: %d, actual: %d", info.desiredNodeNum, info.actualNodeNum),
		})
	}
	return out, nil
}

// componentNodeInfo represents TC component node info for calculating node ready result
type componentNodeInfo struct {
	tcName         string
	componentType  string
	nodeSelector   labels.Selector
	desiredNodeNum int32
	actualNodeNum  int32
}

func (s *AgentService) buildComponentNodeInfos(ctx context.Context, tcs map[string]*v1alpha1.TidbCluster, compTypes []apicommon.ComponentType) ([]*componentNodeInfo, error) {
	logger := tilog.WithContextL(ctx)

	var componentInfos []*componentNodeInfo
	for _, tc := range tcs {
		for _, comp := range tc.AllComponentSpec() {
			compType := comp.MemberType().String()
			if !lo.Contains(compTypes, convertComponentTypeToAPI(compType)) {
				logger.Debug("skip for component", zap.String("comp", compType))
				continue
			}

			info := &componentNodeInfo{}
			info.tcName = tc.Name
			info.componentType = comp.MemberType().String()
			info.nodeSelector, _ = metav1.LabelSelectorAsSelector(metav1.SetAsLabelSelector(comp.NodeSelector()))
			info.desiredNodeNum = getComponentReplicas(tc, comp.MemberType())
			componentInfos = append(componentInfos, info)
		}
	}
	return componentInfos, nil
}

// getNodesByClusterId returns all nodes referenced by this cluster
// NOTE: For nodes which aren't using by TC, only can be retrieved for AWS(EKS). For GCP(GKE), only in use nodes can be retrieved.
func (s *AgentService) getNodesByClusterId(ctx context.Context, clusterId string) (map[string]*corev1.Node, error) {
	logger := tilog.WithContextL(ctx)
	// Since we use shared nodepool for GKE cluster, each node in GKE nodepool doesn't have a label like 'tidbcloud-cluster'.
	// So we should get pods owned by this cluster first, then get nodes by pod.spec.nodeName.
	name2pod, err := s.getPodsByClusterId(ctx, clusterId)
	if err != nil {
		return nil, fmt.Errorf("get pods failed: %w", err)
	}
	logger.Info("pod num", zap.Int("num", len(name2pod)))

	var hostnames []string
	nodeName2ComponentType := make(map[string]string, len(name2pod))
	for _, pod := range name2pod {
		if pod.Spec.NodeName == "" {
			continue
		}

		hostnames = append(hostnames, pod.Spec.NodeName)
		// for TC pods, every pod has its dedicated node, so we can use pod's componentType as the node's componentType
		componentTypeInLabel := pod.Labels[label.ComponentLabelKey]
		// for over provisioning pods, we need to use over provisioning component type
		if componentTypeInLabel == consts.OverProvisioningComponentLabelValue {
			componentTypeInLabel = pod.Labels[consts.OverProvisioningComponentLabelKey]
		}
		nodeName2ComponentType[pod.Spec.NodeName] = componentTypeInLabel
	}
	logger.Info("hostname num", zap.Int("num", len(hostnames)))
	logger.Debug("hostnames", zap.Strings("hostnames", hostnames))
	if len(hostnames) == 0 {
		return nil, nil
	}

	labelSelector, err := metav1.LabelSelectorAsSelector(&metav1.LabelSelector{
		MatchExpressions: []metav1.LabelSelectorRequirement{
			{
				Key:      corev1.LabelHostname,
				Operator: metav1.LabelSelectorOpIn,
				Values:   hostnames,
			},
		},
	})
	if err != nil {
		return nil, fmt.Errorf("build label selector: %w", err)
	}
	var nodeList corev1.NodeList
	if err := s.dataplaneCli.List(ctx, &nodeList, client.MatchingLabelsSelector{Selector: labelSelector}); err != nil {
		return nil, fmt.Errorf("failed to list nodes by pod's hostname: %w", err)
	}

	// list nodes for AWS cluster. only AWS use dedicated NodeGroup/AutoScalingGroup per cluster, component, zone.
	var nodesForAWSCluster corev1.NodeList
	if err := s.dataplaneCli.List(ctx, &nodesForAWSCluster, client.MatchingLabels{"tidb-namespace": buildNamespace(clusterId)}); err != nil {
		return nil, fmt.Errorf("failed to list nodes by node's label: %w", err)
	}
	var nodes []corev1.Node
	nodes = append(nodes, nodeList.Items...)
	nodes = append(nodes, nodesForAWSCluster.Items...)
	out := lo.SliceToMap(nodes, func(item corev1.Node) (string, *corev1.Node) {
		// use pod's componentType as its dedicated node's componentType
		if _, ok := item.Labels[namer.ComponentSelectorKey]; !ok {
			item.Labels[namer.ComponentSelectorKey] = nodeName2ComponentType[item.Name]
		}

		return item.Name, &item
	})
	return out, nil
}

func parseProviderInstanceNameFromNode(no *corev1.Node) string {
	provider, err := getCloudProviderFromNode(no)
	if err != nil {
		return "parse failed"
	}
	switch provider {
	case "aws":
		_, instanceId, err := parseEKSNodeProviderID(no.Spec.ProviderID)
		if err != nil {
			return "parse AWS provider id failed"
		}
		return instanceId
	case "gcp":
		_, _, instanceId, err := parseGKENodeProviderID(no.Spec.ProviderID)
		if err != nil {
			return "parse GCP provider id failed"
		}
		return instanceId
	case "azure":
		// do nothing
		// TODO: add azure control logic
		return ""
	default:
		return fmt.Sprintf("unknown provider: %s", provider)
	}
}

func convertNode(no *corev1.Node) *apiv1.Node {
	out := &apiv1.Node{
		K8SMetadata:          buildK8sMetadata(no),
		ComponentType:        convertComponentTypeToAPI(no.Labels[namer.ComponentSelectorKey]),
		Region:               no.Labels[corev1.LabelTopologyRegion],
		Zone:                 no.Labels[corev1.LabelTopologyZone],
		Hostname:             no.Labels[corev1.LabelHostname],
		ProviderInstanceName: parseProviderInstanceNameFromNode(no),
		ProviderInstanceType: no.Labels[corev1.LabelInstanceTypeStable],
		CpuArch:              no.Labels[corev1.LabelArchStable],
		Ready:                false, // fill later
		ReadyReason:          "",    // fill later
	}

	for _, condition := range no.Status.Conditions {
		if condition.Type == corev1.NodeReady {
			out.Ready = condition.Status == corev1.ConditionTrue
			out.ReadyReason = condition.Reason
		}
	}
	return out
}

func buildOverProvisioningDeploymentLabels(tcName, componentType string) map[string]string {
	return map[string]string{
		consts.InstanceLabelKey:                  tcName,
		consts.ComponentLabelKey:                 consts.OverProvisioningComponentLabelValue,
		consts.OverProvisioningComponentLabelKey: componentType,
	}
}

func getComponentReplicas(tc *v1alpha1.TidbCluster, compType v1alpha1.MemberType) int32 {
	var replicas int32
	switch compType {
	case v1alpha1.PDMemberType:
		replicas = tc.Spec.PD.Replicas
	case v1alpha1.TiKVMemberType:
		replicas = tc.Spec.TiKV.Replicas
	case v1alpha1.TiDBMemberType:
		replicas = tc.Spec.TiDB.Replicas
	case v1alpha1.TiFlashMemberType:
		replicas = tc.Spec.TiFlash.Replicas
	case v1alpha1.TiProxyMemberType:
		replicas = tc.Spec.TiProxy.Replicas
	}
	return replicas
}