package server

import (
	"context"
	"fmt"

	"github.com/samber/lo"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	"k8s.io/apimachinery/pkg/api/resource"
	"k8s.io/apimachinery/pkg/labels"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

func (s *serverImpl) SetHPASpec(
	ctx context.Context,
	sharedPoolId string,
	groupId string,
	componentType entity.ComponentType,
	spec *entity.HorizontalPodAutoscalerSpec,
) error {
	if sharedPoolId == "" {
		return status.Errorf(codes.InvalidArgument, "sharedPoolId is required")
	}
	unlockFunc, err := s.lockTiKVRemoteWorkerGuardedMutation(ctx, sharedPoolId, componentType)
	if err != nil {
		return err
	}
	defer unlockFunc()

	sharedPool, err := s.sharedPoolRepo.Load(ctx, sharedPoolId, false)
	if err != nil {
		return err
	}
	if err := guardSharedPoolMutation(sharedPool, sharedPoolMutationBlockedWhilePaused); err != nil {
		return err
	}
	if componentType == apiv1.ComponentType_COPROCESSOR_WORKER {
		if err := ensureCoprocessorWorkerComponentExists(sharedPool); err != nil {
			return err
		}
	}
	if isTiKVRemoteWorkerGuardedComponentType(componentType) {
		if err := s.mergeSharedPoolStatusFromDagTask(ctx, sharedPool); err != nil {
			return status.Errorf(codes.Internal, "failed to merge sharedPool status from dag task: %s", err)
		}
		if err := s.guardTiKVRemoteWorkersTopologyStable(ctx, sharedPool); err != nil {
			return err
		}
	}
	remoteCli, err := s.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
	if err != nil {
		return status.Errorf(codes.Internal, "failed to get remote client for shared pool: %s", err)
	}
	namespace := "tidb" + sharedPool.GetId()
	hpas := new(autoscalingv2.HorizontalPodAutoscalerList)
	switch componentType {
	case apiv1.ComponentType_TIDB:
		if err := remoteCli.List(ctx, hpas, &client.ListOptions{
			Namespace: namespace,
			LabelSelector: labels.SelectorFromSet(map[string]string{
				schemav1.LabelPingcapGroupKey: fmt.Sprintf("db-%s", groupId),
				schemav1.LabelComponentKey:    "tidb",
			}),
		}); err != nil {
			return status.Errorf(codes.Internal, "failed to list tidb hpas: %s", err)
		}
	case apiv1.ComponentType_TIDB_WORKER:
		if err := remoteCli.List(ctx, hpas, &client.ListOptions{
			Namespace: namespace,
			LabelSelector: labels.SelectorFromSet(map[string]string{
				schemav1.LabelComponentKey: "worker-tidb",
			}),
		}); err != nil {
			return status.Errorf(codes.Internal, "failed to list tidb-worker hpas: %s", err)
		}
	case apiv1.ComponentType_TIKV_WORKER:
		if err := remoteCli.List(ctx, hpas, &client.ListOptions{
			Namespace: namespace,
			LabelSelector: labels.SelectorFromSet(map[string]string{
				schemav1.LabelComponentKey: entity.TiKVWorkerType,
			}),
		}); err != nil {
			return status.Errorf(codes.Internal, "failed to list tikv-worker hpas: %s", err)
		}
	case apiv1.ComponentType_COPROCESSOR_WORKER:
		if err := remoteCli.List(ctx, hpas, &client.ListOptions{
			Namespace: namespace,
			LabelSelector: labels.SelectorFromSet(map[string]string{
				schemav1.LabelComponentKey: entity.CoprocessorWorkerType,
			}),
		}); err != nil {
			return status.Errorf(codes.Internal, "failed to list coprocessor-worker hpas: %s", err)
		}
	default:
		return status.Errorf(codes.Unimplemented, "%s does not support hpa", componentType)
	}
	if len(hpas.Items) == 0 {
		return status.Errorf(codes.Internal, "%s has no hpa", componentType)
	}
	hpa := &hpas.Items[0]

	changed, err := patchHPASpec(hpa, spec)
	if err != nil {
		return err
	}
	if changed {
		if err := remoteCli.Update(ctx, hpa); err != nil {
			return status.Errorf(codes.Internal, "%s", err.Error())
		}
	}
	return nil
}

func patchMetricTarget(curr *autoscalingv2.MetricTarget, next *entity.MetricTarget) (bool, error) {
	changed := false
	if next.Value != nil && curr.Value != nil {
		quantity, err := resource.ParseQuantity(next.GetValue())
		if err != nil {
			return false, err
		}
		curr.Value = &quantity
		changed = true
	}
	if next.AverageValue != nil && curr.AverageValue != nil {
		quantity, err := resource.ParseQuantity(next.GetAverageValue())
		if err != nil {
			return false, err
		}
		curr.AverageValue = &quantity
		changed = true
	}
	if next.AverageUtilization != nil && curr.AverageUtilization != nil {
		*curr.AverageUtilization = next.GetAverageUtilization()
		changed = true
	}
	return changed, nil
}

func pathcHPAScalingRules(curr *autoscalingv2.HPAScalingRules, next *entity.HPAScalingRules) bool {
	changed := false
	if value := next.StabilizationWindowSeconds; value != nil &&
		*value != lo.FromPtr(curr.StabilizationWindowSeconds) {
		curr.StabilizationWindowSeconds = lo.ToPtr(next.GetStabilizationWindowSeconds())
		changed = true
	}
	// slice can only be replace entirly
	if len(next.GetPolicies()) != 0 {
		curr.Policies = toV2HPAScalingPolicy(next.GetPolicies())
	}
	return changed
}

func patchHPASpec(hpa *autoscalingv2.HorizontalPodAutoscaler, spec *entity.HorizontalPodAutoscalerSpec) (bool, error) {
	if spec == nil {
		return false, nil
	}
	key := client.ObjectKeyFromObject(hpa)
	changed := false
	for _, metricSpec := range spec.GetMetrics() {
		for _, curr := range hpa.Spec.Metrics {
			if metricSource := metricSpec.GetObject(); metricSource != nil && curr.Object != nil {
				if metricSource.GetMetric().GetName() != curr.Object.Metric.Name {
					continue
				}
				target := metricSource.GetTarget()
				if target == nil {
					continue
				}
				currTarget := &curr.Object.Target
				if targetChanged, err := patchMetricTarget(currTarget, target); err != nil {
					return false, err
				} else {
					changed = targetChanged || changed
				}
			}
			if metricSource := metricSpec.Pods; metricSource != nil && curr.Pods != nil {
				if metricSource.GetMetric().GetName() != curr.Pods.Metric.Name {
					continue
				}
				target := metricSource.GetTarget()
				if target == nil {
					continue
				}
				currTarget := &curr.Pods.Target
				if targetChanged, err := patchMetricTarget(currTarget, target); err != nil {
					return false, err
				} else {
					changed = targetChanged || changed
				}
			}
			if metricSource := metricSpec.Resource; metricSource != nil && curr.Resource != nil {
				if metricSource.GetName() != curr.Resource.Name.String() {
					continue
				}
				target := metricSource.GetTarget()
				if target == nil {
					continue
				}
				currTarget := &curr.Resource.Target
				if targetChanged, err := patchMetricTarget(currTarget, target); err != nil {
					return false, err
				} else {
					changed = targetChanged || changed
				}
			}
			if metricSource := metricSpec.ContainerResource; metricSource != nil && curr.ContainerResource != nil {
				if metricSource.GetName() != curr.Resource.Name.String() {
					continue
				}
				if metricSource.GetContainer() != curr.ContainerResource.Container {
					continue
				}
				target := metricSource.GetTarget()
				if target == nil {
					continue
				}
				currTarget := &curr.ContainerResource.Target
				if targetChanged, err := patchMetricTarget(currTarget, target); err != nil {
					return false, err
				} else {
					changed = targetChanged || changed
				}
			}
			if metricSource := metricSpec.External; metricSource != nil && curr.External != nil {
				if metricSource.GetMetric().GetName() != curr.External.Metric.Name {
					continue
				}
				target := metricSource.GetTarget()
				if target == nil {
					continue
				}
				currTarget := &curr.External.Target
				if targetChanged, err := patchMetricTarget(currTarget, target); err != nil {
					return false, err
				} else {
					changed = targetChanged || changed
				}
			}
		}
	}

	if behavior := spec.GetBehavior(); behavior != nil {
		if hpa.Spec.Behavior == nil {
			return false, status.Errorf(codes.Internal, "data corrupted, hpa %s has nil behavior", key)
		}
		// the hpa get from dataplane k8s should always has behavior.scaleUp and behavior.scaleDown
		if scaleUp := behavior.GetScaleUp(); scaleUp != nil {
			if hpa.Spec.Behavior.ScaleUp == nil {
				return false, status.Errorf(codes.Internal, "data corrupted, hpa %s has nil scaleUp", key)
			}
			rulesChanged := pathcHPAScalingRules(hpa.Spec.Behavior.ScaleUp, scaleUp)
			changed = rulesChanged || changed
		}

		if scaleDown := behavior.GetScaleDown(); scaleDown != nil {
			if hpa.Spec.Behavior.ScaleDown == nil {
				return false, status.Errorf(codes.Internal, "data corrupted, hpa %s has nil scaleDown", key)
			}
			rulesChanged := pathcHPAScalingRules(hpa.Spec.Behavior.ScaleDown, scaleDown)
			changed = rulesChanged || changed
		}
	}
	return changed, nil
}

func toV2HPAScalingPolicy(in []*entity.HPAScalingPolicy) []autoscalingv2.HPAScalingPolicy {
	out := make([]autoscalingv2.HPAScalingPolicy, 0, len(in))
	for _, item := range in {
		switch item.Type {
		case string(autoscalingv2.PodsScalingPolicy):
			out = append(out, autoscalingv2.HPAScalingPolicy{
				Type:          autoscalingv2.PodsScalingPolicy,
				Value:         item.Value,
				PeriodSeconds: item.PeriodSeconds,
			})
		case string(autoscalingv2.PercentScalingPolicy):
			out = append(out, autoscalingv2.HPAScalingPolicy{
				Type:          autoscalingv2.PercentScalingPolicy,
				Value:         item.Value,
				PeriodSeconds: item.PeriodSeconds,
			})
		default:
			continue
		}
	}
	return out
}