package tidbautoscaling

import (
	"fmt"
	"math"
	"time"

	"github.com/pingcap/errors"
	serverlessv1alpha1 "github.com/tidbcloud/aws-shared-provider/apis/serverless/v1alpha1"
	"github.com/tidbcloud/aws-shared-provider/pkg/metrics"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
)

// Float3 wraps a float64 to format to 3 decimal places when printed
type Float3 float64

// String implements fmt.Stringer to format Float3 with three decimals
func (f Float3) String() string {
	return fmt.Sprintf("%.3f", float64(f))
}

type detail struct {
	MetricName       string
	CurAvgValue      Float3
	CurScaleFlapping Float3
}

func (ta *TidbAutoscaling) scale(
	tokenMetricMap map[string]tokenMetric,
	taConfig *serverlessv1alpha1.TidbAutoscaling,
	tidbMaxReplicas int,
	disableScaleDown map[string]bool,
) {
	scaleRecordMap := make(map[string]scaleRecord, len(ta.scaleRecordMap))
	defer func() {
		ta.scaleRecordMap = scaleRecordMap
	}()

	for token, metric := range tokenMetricMap {
		logger := ta.WithValues("token", token)

		if scale, ok := ta.scaleRecordMap[token]; ok {
			scaleRecordMap[token] = scale
		}

		c, err := ta.getInfraObjWithToken(utils.TrimToken(token))
		if err != nil {
			logger.Error(err, "get infra obj with token failed")
			continue
		}
		if _, ok := c.(*cloudv1.Branch); ok {
			logger.V(1).Info("skip branch for autoscaling")
			continue
		}
		specForProvider, err := types.InfraServerlessSpecForProvider(c)
		if err != nil {
			logger.Error(err, "get specForProvider failed")
			continue
		}
		logger = logger.WithValues("cluster_name", specForProvider.ClusterName)

		if err := validateScale(metric); err != nil {
			logger.V(1).Info("validate scale", "skip", err.Error())
			continue
		}

		logger.V(3).Info("try to scale", "currentReplicas", metric.replicas)
		//nolint:govet // tidbMaxReplicas should be shadowed here
		tidbMinReplicas, tidbMaxReplicas := types.GetDefaultTiDBMinReplicas(specForProvider), tidbMaxReplicas
		if specForProvider.TidbMaxReplicas > 0 {
			tidbMaxReplicas = int(specForProvider.TidbMaxReplicas)
			logger.V(1).Info("use infra spec.forProvider", "tidbMaxReplicas", specForProvider.TidbMaxReplicas)
		}
		if specForProvider.TidbMinReplicas > 0 {
			tidbMinReplicas = int(specForProvider.TidbMinReplicas)
			logger.V(1).Info("use infra spec.forProvider", "tidbMinReplicas", specForProvider.TidbMinReplicas)
		}
		if tidbMinReplicas > tidbMaxReplicas {
			logger.Info("tidbMinReplicas > tidbMaxReplicas, reset tidbMinReplicas to 1",
				"tidbMinReplicas", tidbMinReplicas, "tidbMaxReplicas", tidbMaxReplicas)
			tidbMinReplicas = types.GetDefaultTiDBMinReplicas(specForProvider)
		}
		rgName := utils.TrimResourceGroup(token)
		if rgName != "" {
			tidbMinReplicas = types.TidbResourceGroupMinReplicas
		}

		targetReplicas, extra := calcTargetReplicas(metric, taConfig)
		if targetReplicas < tidbMinReplicas {
			logger.Info("targetReplicas < tidbMinReplicas",
				"targetReplicas", targetReplicas, "tidbMinReplicas", tidbMinReplicas)
			targetReplicas = tidbMinReplicas
		}
		if targetReplicas > tidbMaxReplicas {
			logger.Info("targetReplicas > tidbMaxReplicas",
				"targetReplicas", targetReplicas, "tidbMaxReplicas", tidbMaxReplicas)
			targetReplicas = tidbMaxReplicas
		}
		if targetReplicas > metric.replicas {
			nowTimestamp := time.Now().Unix()
			preScale, ok := scaleRecordMap[token]
			if !ok || preScale.direction != UP {
				preScale = scaleRecord{direction: UP, timestamp: nowTimestamp}
			}
			stableSeconds := nowTimestamp - preScale.timestamp
			if stableSeconds < taConfig.ScaleUpStabilizationWindowSeconds {
				scaleRecordMap[token] = preScale
				logger.V(1).Info("in scale up stabilization window", "stableSeconds", stableSeconds,
					"ScaleUpStabilizationWindowSeconds", taConfig.ScaleUpStabilizationWindowSeconds,
					"targetReplicas", targetReplicas, "currentReplicas", metric.replicas)
				continue
			}
			logger.Info("scale up tidb", "targetReplicas", targetReplicas, "currentReplicas", metric.replicas, "extra", extra)
		}
		if targetReplicas < metric.replicas {
			if specForProvider.ScaleDownOneEachTime {
				targetReplicas = metric.replicas - 1
			}
			nowTimestamp := time.Now().Unix()
			preScale, ok := scaleRecordMap[token]
			if !ok || preScale.direction != DOWN {
				preScale = scaleRecord{direction: DOWN, timestamp: nowTimestamp}
			}
			stableSeconds := nowTimestamp - preScale.timestamp
			scaleDownWindow := getScaleDownWindowSeconds(specForProvider, taConfig)
			if stableSeconds < scaleDownWindow {
				scaleRecordMap[token] = preScale
				logger.V(1).Info("in scale down stabilization window",
					"stableSeconds", stableSeconds,
					"ScaleDownStabilizationWindowSeconds", scaleDownWindow,
					"targetReplicas", targetReplicas, "currentReplicas", metric.replicas)
				continue
			}
			if disableScaleDown[token] {
				logger.V(1).Info("scale down is disabled by check",
					"targetReplicas", targetReplicas, "currentReplicas", metric.replicas)
				continue
			}
			if specForProvider.TidbPoolChanging {
				logger.V(1).Info("tidb pool is changing, skip scale down",
					"targetReplicas", targetReplicas, "currentReplicas", metric.replicas)
				continue
			}
			logger.Info("scale down tidb", "targetReplicas", targetReplicas, "currentReplicas", metric.replicas, "extra", extra)
		}
		if targetReplicas == metric.replicas {
			logger.V(3).Info("the replicas not change", "replicas", targetReplicas)
			if _, ok := scaleRecordMap[token]; ok {
				scaleRecordMap[token] = scaleRecord{direction: HOLD, timestamp: time.Now().Unix()}
			}
		} else {
			if err := ta.patchTidbReplicas(c, rgName, targetReplicas); err != nil {
				logger.Error(err, "patch tidb replicas failed")
				continue
			}

			if targetReplicas > metric.replicas {
				scaleRecordMap[token] = scaleRecord{direction: UP, timestamp: time.Now().Unix()}
				metrics.ScaleCounter.WithLabelValues("up", ta.tidbPoolNamespace).Inc()
			} else {
				scaleRecordMap[token] = scaleRecord{direction: DOWN, timestamp: time.Now().Unix()}
				metrics.ScaleCounter.WithLabelValues("down", ta.tidbPoolNamespace).Inc()
			}
		}
	}
}

func getScaleDownWindowSeconds(specForProvider *types.InfraServerlessSpec, ta *serverlessv1alpha1.TidbAutoscaling) int64 {
	if specForProvider.ScaleDownStabilizationWindowSeconds != nil {
		return *specForProvider.ScaleDownStabilizationWindowSeconds
	}
	if specForProvider.TiDBMinRCU > 0 {
		return ta.ScaleDownEssentialStabilizationWindowSeconds
	}
	return ta.ScaleDownStabilizationWindowSeconds
}

func calcTargetReplicas(tMetric tokenMetric, ta *serverlessv1alpha1.TidbAutoscaling) (targetReplicas int, extra detail) {
	for i, metric := range ta.Metrics {
		var sum float64
		for _, podMetric := range tMetric.podMetrics {
			sum += podMetric[i].value
		}
		curAvgValue := sum / float64(tMetric.replicas)
		r, f := calcReplicas(tMetric.replicas, metric, curAvgValue)
		if r > targetReplicas {
			targetReplicas = r
			extra = detail{MetricName: metric.Name, CurAvgValue: Float3(curAvgValue), CurScaleFlapping: Float3(f)}
		}
	}
	return
}

func calcReplicas(curReplicas int, metric serverlessv1alpha1.Metric, curAvgValue float64) (int, float64) {
	curScaleFlapping := math.Abs(metric.AvgValue-curAvgValue) / metric.AvgValue
	if curScaleFlapping < metric.ScaleFlapping {
		return curReplicas, curScaleFlapping
	}
	return int(math.Ceil(float64(curReplicas) * curAvgValue / metric.AvgValue)), curScaleFlapping
}

func validateScale(metric tokenMetric) error {
	if metric.replicas == 0 {
		return errors.New("replicas is 0")
	}
	if metric.replicas != len(metric.podMetrics) {
		return errors.New("replicas not equal to actual")
	}
	for pod, podMetric := range metric.podMetrics {
		if podMetric == nil {
			return errors.Errorf("pod metric is nil, pod name: %s", pod)
		}
		for i := range podMetric {
			if !podMetric[i].isValid {
				return errors.Errorf("pod metric is invalid, pod name: %s, metric index: %d", pod, i)
			}
		}
	}
	return nil
}