package dataplane

import (
	"context"
	"fmt"
	"sort"
	"strconv"

	ackv1alpha1 "github.com/crossplane-contrib/provider-upjet-alibabacloud/apis/ack/v1alpha1"
	"github.com/pingcap/log"
	eksv1beta1 "github.com/upbound/provider-aws/apis/eks/v1beta1"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	dmpbv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
)

var cloudProviders = map[string]string{
	consts.ProviderTypeEKS: "aws",
	consts.ProviderTypeACK: "alicloud",
	consts.ProviderTypeAKS: "azure",
	consts.ProviderTypeGKE: "gcp",
}

func (s *DataPlaneService) ListK8SInformation(
	ctx context.Context, req *dmpbv1.ListK8SInformationRequest) (*dmpbv1.K8SInformationList, error) {
	version := req.GetVersion()
	if version == "" {
		log.Error("version is required for ListK8SInformation")
		return nil, bizerr.New(bizerr.ParameterInvalid, "version is required")
	}

	appBoxList := &v1alpha1.APPBoxList{}
	if err := s.appBoxRW.List(ctx, appBoxList); err != nil {
		log.Error("failed to list appboxes", zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to list appboxes", err)
	}
	dataPlaneList := &v1alpha1.DataPlaneList{}
	if err := s.dataPlaneRW.List(ctx, dataPlaneList); err != nil {
		log.Error("failed to list dataplanes", zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to list dataplanes", err)
	}
	appBoxToDataPlane := make(map[string]*v1alpha1.DataPlane)
	for i := range dataPlaneList.Items {
		dp := &dataPlaneList.Items[i]
		if dp.Status.AppBox != nil && dp.Status.AppBox.Name != "" {
			key := dp.Status.AppBox.Namespace + "/" + dp.Status.AppBox.Name
			appBoxToDataPlane[key] = dp
		}
	}

	k8sInfos := make([]*dmpbv1.K8SInformation, 0, len(appBoxList.Items))
	for i := range appBoxList.Items {
		appBox := &appBoxList.Items[i]
		// Filter by version
		// NOTE: we can't use FieldsSelector as NO controller-runtime Manager is used in the dataplane Server
		if appBox.Spec.KubernetesConfig.Version != version {
			continue
		}
		var dataplane *v1alpha1.DataPlane
		key := appBox.Namespace + "/" + appBox.Name
		if dp, ok := appBoxToDataPlane[key]; ok {
			dataplane = dp
		}
		if dataplane == nil {
			log.Error("dataplane not found for appbox", zap.String("appbox", key))
			return nil, bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("dataplane not found for appbox %s", key))
		}

		// NOTE: do NOT fill AdminNodes for List API
		k8sInfo, err := s.appBox2K8sInfo(ctx, appBox, dataplane, false)
		if err != nil {
			log.Error("failed to convert appbox to k8sinfo", zap.String("appbox", key), zap.Error(err))
			return nil, err
		}
		k8sInfos = append(k8sInfos, k8sInfo)
	}

	return &dmpbv1.K8SInformationList{
		K8SInfos: k8sInfos,
		Total:    uint32(len(k8sInfos)),
	}, nil
}

func (s *DataPlaneService) GetK8SInformation(
	ctx context.Context, req *dmpbv1.GetK8SInformationRequest) (*dmpbv1.K8SInformation, error) {
	dataPlaneId := req.GetDataPlaneId()
	if dataPlaneId == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "data_plane_id is required")
	}

	var dataPlane v1alpha1.DataPlane
	if err := s.dataPlaneRW.Get(ctx, client.ObjectKey{
		Namespace: s.config.Namespace,
		Name:      dataPlaneId, // dataPlaneId is the name of the DataPlane CR
	}, &dataPlane); err != nil {
		log.Error("failed to get dataplane", zap.String("data_plane_id", dataPlaneId), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to get dataplane", err)
	}
	if dataPlane.Status.AppBox == nil {
		log.Error("dataplane does not have appbox reference", zap.String("data_plane_id", dataPlaneId))
		return nil, bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("dataplane %s does not have appbox reference", dataPlaneId))
	}

	var appBox v1alpha1.APPBox
	if err := s.appBoxRW.Get(ctx, client.ObjectKey{
		Namespace: dataPlane.Status.AppBox.Namespace,
		Name:      dataPlane.Status.AppBox.Name,
	}, &appBox); err != nil {
		log.Error("failed to get appbox", zap.String("name", dataPlane.Status.AppBox.Name), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to get appbox", err)
	}

	return s.appBox2K8sInfo(ctx, &appBox, &dataPlane, true)
}

func (s *DataPlaneService) UpgradeDataPlane(
	ctx context.Context, req *dmpbv1.UpgradeDataPlaneRequest,
) (*emptypb.Empty, error) {
	dataPlaneId := req.GetDataPlaneId()
	k8sVersion := req.GetK8SVersion()
	if dataPlaneId == "" || k8sVersion == "" {
		log.Error("data_plane_id and k8s_version are required for UpgradeK8S")
		return nil, bizerr.New(bizerr.ParameterInvalid, "data_plane_id and k8s_version are required")
	}

	var dataPlane v1alpha1.DataPlane
	if err := s.dataPlaneRW.Get(ctx, client.ObjectKey{
		Namespace: s.config.Namespace,
		Name:      dataPlaneId,
	}, &dataPlane); err != nil {
		log.Error("failed to get dataplane", zap.String("data_plane_id", dataPlaneId), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to get dataplane", err)
	}
	if dataPlane.Status.AppBox == nil {
		log.Error("dataplane does not have appbox reference", zap.String("data_plane_id", dataPlaneId))
		return nil, bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("dataplane %s does not have appbox reference", dataPlaneId))
	}

	var appBox v1alpha1.APPBox
	if err := s.appBoxRW.Get(ctx, client.ObjectKey{
		Namespace: dataPlane.Status.AppBox.Namespace,
		Name:      dataPlane.Status.AppBox.Name,
	}, &appBox); err != nil {
		log.Error("failed to get appbox", zap.String("name", dataPlane.Status.AppBox.Name), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to get appbox", err)
	}

	// check if the version is already the desired version
	if appBox.Spec.KubernetesConfig.Version == k8sVersion {
		log.Info("appbox already at the desired version", zap.String("data_plane_id", dataPlaneId),
			zap.String("appbox", appBox.Name), zap.String("k8s_version", k8sVersion))
		return &emptypb.Empty{}, nil
	}

	// upgrade the version in the appbox spec, the controller will handle the rest
	// this includes:
	// - upgrading the EKS/ACK control plane
	// - upgrading the addons of the cluster
	// - upgrading the admin node group/node pool
	// - creating new node groups for EKS
	appBox.Spec.KubernetesConfig.Version = k8sVersion
	if err := s.appBoxRW.Update(ctx, &appBox); err != nil {
		log.Error("failed to update appbox for upgrade", zap.String("dataplane_id", dataPlaneId),
			zap.String("appbox", appBox.Name), zap.Error(err))
		return nil, bizerr.Wrap(bizerr.K8SClientError, "failed to update appbox for upgrade", err)
	}

	return &emptypb.Empty{}, nil
}

func (s *DataPlaneService) appBox2K8sInfo(
	ctx context.Context, appBox *v1alpha1.APPBox, dataplane *v1alpha1.DataPlane,
	getDetails bool) (*dmpbv1.K8SInformation, error) {
	var maxSkewVersion uint32
	switch appBox.Spec.Provider {
	case consts.ProviderTypeEKS:
		// 3 from EKS v1.28+, refer to:
		// https://docs.aws.amazon.com/eks/latest/best-practices/cluster-upgrades.html#node-compatibility
		maxSkewVersion = 3
	case consts.ProviderTypeACK:
		// for ACK, need to upgrade version step by step
		maxSkewVersion = 1
	default:
		return nil, bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("unsupported provider: %s", appBox.Spec.Provider))
	}

	k8sInfo := &dmpbv1.K8SInformation{
		DataPlaneId:    dataplane.Name, // dataplane.Name is the dataplane ID
		CloudProvider:  cloudProviders[appBox.Spec.Provider],
		CloudAccount:   dataplane.Spec.AccountRef.AccountID,
		Region:         appBox.Spec.Region,
		TenantId:       appBox.Labels[consts.TenantLabelKey],
		ResourcePoolId: appBox.Labels[consts.SharedPoolLabelKey],
		AppboxName:     appBox.Name,
		TargetVersion:  appBox.Spec.KubernetesConfig.Version,
		MaxSkewVersion: maxSkewVersion,
		ControlPlane: &dmpbv1.K8SControlPlane{
			Name: appBox.Status.KubernetesClusterName,
		},
	}

	if err := s.fillK8sVersion(ctx, appBox, k8sInfo); err != nil {
		return nil, err
	}

	// Calculate expected worker node pool count
	switch appBox.Spec.Provider {
	case consts.ProviderTypeEKS:
		// For PerZonePerInstanceType mode: nodePools × zones
		nodePools := s.config.ControllerConfig.APPBoxConfig.NodePools
		zones := append([]string(nil), dataplane.Spec.NodeZones...)
		if len(zones) == 0 {
			zones = append([]string(nil), dataplane.Spec.AvailabilityZones...)
		}
		k8sInfo.ExpectedWorkerNodePoolCount = uint32(len(nodePools) * len(zones))
	case consts.ProviderTypeACK:
		// For ACK, a node pool is across all zones
		k8sInfo.ExpectedWorkerNodePoolCount = uint32(len(s.config.ControllerConfig.APPBoxConfig.NodePools))
	}

	if getDetails {
		if err := s.fillAddonsInfo(ctx, appBox, k8sInfo); err != nil {
			return nil, err
		}
		if err := s.fillNodesInfo(ctx, appBox, k8sInfo); err != nil {
			return nil, err
		}
	}

	return k8sInfo, nil
}

func (s *DataPlaneService) fillK8sVersion(
	ctx context.Context, appBox *v1alpha1.APPBox, k8sInfo *dmpbv1.K8SInformation) error {
	switch appBox.Spec.Provider {
	case consts.ProviderTypeEKS:
		var eksCluster eksv1beta1.Cluster
		if err := s.localCli.Get(ctx, client.ObjectKey{Name: k8sInfo.ControlPlane.Name}, &eksCluster); err != nil {
			log.Error("failed to get eks cluster", zap.String("appbox", appBox.Name), zap.String("cluster", k8sInfo.ControlPlane.Name), zap.Error(err))
			return bizerr.Wrap(bizerr.K8SClientError, "failed to get eks cluster", err)
		}
		k8sInfo.ControlPlane.Version = ptr.Deref(eksCluster.Status.AtProvider.Version, "")

	case consts.ProviderTypeACK:
		var ackCluster ackv1alpha1.ManagedKubernetes
		if err := s.localCli.Get(ctx, client.ObjectKey{Name: k8sInfo.ControlPlane.Name}, &ackCluster); err != nil {
			log.Error("failed to get ack cluster", zap.String("appbox", appBox.Name), zap.String("cluster", k8sInfo.ControlPlane.Name), zap.Error(err))
			return bizerr.Wrap(bizerr.K8SClientError, "failed to get ack cluster", err)
		}
		k8sInfo.ControlPlane.Version = ptr.Deref(ackCluster.Status.AtProvider.Version, "")

	default:
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("unsupported provider: %s", appBox.Spec.Provider))
	}
	return nil
}

func (s *DataPlaneService) fillAddonsInfo(
	ctx context.Context, appBox *v1alpha1.APPBox, k8sInfo *dmpbv1.K8SInformation) error {
	// Get addon template config
	addonTemplate, exists := s.tmplConfig.GetAddonTemplate(appBox.Spec.KubernetesConfig.AddonTemplateID)
	if !exists {
		log.Error("addon template not found", zap.String("addon_template_id", appBox.Spec.KubernetesConfig.AddonTemplateID))
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("addon template %s not found", appBox.Spec.KubernetesConfig.AddonTemplateID))
	}

	targetK8sVersion := appBox.Spec.KubernetesConfig.Version

	switch appBox.Spec.Provider {
	case consts.ProviderTypeEKS:
		var addonList eksv1beta1.AddonList
		if err := s.localCli.List(ctx, &addonList, utils.GetAddonsLabelSelector(appBox)); err != nil {
			log.Error("failed to list eks addons", zap.String("appbox", appBox.Name), zap.String("dataplane", k8sInfo.DataPlaneId), zap.Error(err))
			return bizerr.Wrap(bizerr.K8SClientError, "failed to list eks addons", err)
		}
		k8sInfo.Addons = eksAddons2PB(addonList, addonTemplate.EKS, targetK8sVersion)

	case consts.ProviderTypeACK:
		var addonList ackv1alpha1.KubernetesAddonList
		if err := s.localCli.List(ctx, &addonList, utils.GetAddonsLabelSelector(appBox)); err != nil {
			log.Error("failed to list ack addons", zap.String("appbox", appBox.Name), zap.String("dataplane", k8sInfo.DataPlaneId), zap.Error(err))
			return bizerr.Wrap(bizerr.K8SClientError, "failed to list ack addons", err)
		}
		k8sInfo.Addons = ackAddons2PB(addonList, addonTemplate.ACK, targetK8sVersion)

	default:
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("unsupported provider: %s", appBox.Spec.Provider))
	}
	return nil
}

func (s *DataPlaneService) fillNodesInfo(
	ctx context.Context, appBox *v1alpha1.APPBox, k8sInfo *dmpbv1.K8SInformation) error {
	switch appBox.Spec.Provider {
	case consts.ProviderTypeEKS:
		var ngList eksv1beta1.NodeGroupList
		if err := s.localCli.List(ctx, &ngList, utils.GetNodePoolsLabelSelector(k8sInfo.DataPlaneId)); err != nil {
			log.Error("failed to list eks nodegroups", zap.String("appbox", appBox.Name), zap.String("dataplane", k8sInfo.DataPlaneId), zap.Error(err))
			return bizerr.Wrap(bizerr.K8SClientError, "failed to list eks nodegroups", err)
		}
		k8sInfo.NodePools = eksManagedNodeGroups2PB(ngList)

	case consts.ProviderTypeACK:
		var npList ackv1alpha1.KubernetesNodePoolList
		if err := s.localCli.List(ctx, &npList, utils.GetNodePoolsLabelSelector(k8sInfo.DataPlaneId)); err != nil {
			log.Error("failed to list ack nodepools", zap.String("appbox", appBox.Name), zap.String("dataplane", k8sInfo.DataPlaneId), zap.Error(err))
			return bizerr.Wrap(bizerr.K8SClientError, "failed to list ack nodepools", err)
		}
		k8sInfo.NodePools = ackManagedNodePools2PB(npList)

	default:
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("unsupported provider: %s", appBox.Spec.Provider))
	}
	return nil
}

func eksManagedNodeGroups2PB(ngList eksv1beta1.NodeGroupList) []*dmpbv1.K8SNodePool {
	ret := make([]*dmpbv1.K8SNodePool, 0, len(ngList.Items))
	for _, item := range ngList.Items {
		ng := &dmpbv1.K8SNodePool{
			Kind:     dmpbv1.K8SNodePool_WORKER_NODES,
			Name:     item.Name,
			Version:  ptr.Deref(item.Status.AtProvider.Version, ""),
			Deleting: item.DeletionTimestamp != nil,
			Ready:    ptr.Deref(item.Status.AtProvider.Status, "") == "ACTIVE",
		}
		// Admin (shared) and dedicated o11y pools are APPBox platform pools: in-place versioned,
		// not blue/green workers. Misclassifying o11y as WORKER inflates ExpectedWorkerNodePoolCount checks.
		if comp := item.Spec.ForProvider.Labels[consts.LabelKeyComponent]; comp != nil && consts.IsPlatformManagedNodeComponent(*comp) {
			ng.Kind = dmpbv1.K8SNodePool_ADMIN_NODES
		}
		if len(item.Status.AtProvider.ScalingConfig) > 0 && item.Status.AtProvider.ScalingConfig[0].DesiredSize != nil {
			ng.Count = uint32(*item.Status.AtProvider.ScalingConfig[0].DesiredSize)
		}
		ret = append(ret, ng)
	}
	// Sort by Kind (ADMIN_NODES first) then by Name
	sort.Slice(ret, func(i, j int) bool {
		if ret[i].Kind != ret[j].Kind {
			return ret[i].Kind < ret[j].Kind
		}
		return ret[i].Name < ret[j].Name
	})
	return ret
}

func ackManagedNodePools2PB(ngList ackv1alpha1.KubernetesNodePoolList) []*dmpbv1.K8SNodePool {
	ret := make([]*dmpbv1.K8SNodePool, 0, len(ngList.Items))
	for _, item := range ngList.Items {
		ng := &dmpbv1.K8SNodePool{
			Kind:     dmpbv1.K8SNodePool_WORKER_NODES,
			Name:     item.Name,
			Version:  "", // no version info in ACK node pool status
			Deleting: item.DeletionTimestamp != nil,
			Ready:    true, // ACK does not expose NodePool status; set Ready to true as ACK does in-place upgrades
		}
		for _, lab := range item.Spec.ForProvider.Labels {
			if lab.Key != nil && lab.Value != nil && *lab.Key == consts.LabelKeyComponent && consts.IsPlatformManagedNodeComponent(*lab.Value) {
				ng.Kind = dmpbv1.K8SNodePool_ADMIN_NODES
				break
			}
		}
		if item.Status.AtProvider.DesiredSize != nil {
			if size, err := strconv.Atoi(*item.Status.AtProvider.DesiredSize); err == nil {
				ng.Count = uint32(size)
			}
		}
		ret = append(ret, ng)
	}
	// Sort by Kind (ADMIN_NODES first) then by Name
	sort.Slice(ret, func(i, j int) bool {
		if ret[i].Kind != ret[j].Kind {
			return ret[i].Kind < ret[j].Kind
		}
		return ret[i].Name < ret[j].Name
	})
	return ret
}

func eksAddons2PB(addonList eksv1beta1.AddonList, addonConfigMap map[string][]rsconfig.AddonConfig, targetK8sVersion string) []*dmpbv1.K8SAddon {
	ret := make([]*dmpbv1.K8SAddon, 0, len(addonList.Items))

	addonTargetVersions := make(map[string]string)
	if addonConfigs, ok := addonConfigMap[targetK8sVersion]; ok {
		for _, conf := range addonConfigs {
			addonTargetVersions[conf.AddonName] = conf.AddonVersion
		}
	}

	for _, item := range addonList.Items {
		addonName := ptr.Deref(item.Spec.ForProvider.AddonName, "")
		addon := &dmpbv1.K8SAddon{
			Name:          addonName,
			Version:       ptr.Deref(item.Status.AtProvider.AddonVersion, ""),
			TargetVersion: addonTargetVersions[addonName], // Read from template config
			K8SVersion:    targetK8sVersion,
		}
		ret = append(ret, addon)
	}
	sort.Slice(ret, func(i, j int) bool {
		return ret[i].Name < ret[j].Name
	})
	return ret
}

func ackAddons2PB(addonList ackv1alpha1.KubernetesAddonList, addonConfigMap map[string][]rsconfig.AddonConfig, targetK8sVersion string) []*dmpbv1.K8SAddon {
	ret := make([]*dmpbv1.K8SAddon, 0, len(addonList.Items))

	addonTargetVersions := make(map[string]string)
	if addonConfigs, ok := addonConfigMap[targetK8sVersion]; ok {
		for _, conf := range addonConfigs {
			addonTargetVersions[conf.AddonName] = conf.AddonVersion
		}
	}

	for _, item := range addonList.Items {
		addonName := ptr.Deref(item.Spec.ForProvider.Name, "")
		addon := &dmpbv1.K8SAddon{
			Name:          addonName,
			Version:       ptr.Deref(item.Status.AtProvider.Version, ""),
			TargetVersion: addonTargetVersions[addonName], // Read from template config
			K8SVersion:    targetK8sVersion,
		}
		ret = append(ret, addon)
	}
	sort.Slice(ret, func(i, j int) bool {
		return ret[i].Name < ret[j].Name
	})
	return ret
}