package configupdate

import (
	"context"
	"errors"
	"fmt"
	"io"
	"math"
	"net/url"
	"strconv"
	"strings"
	"time"

	. "github.com/onsi/gomega"
	pdclient "github.com/pingcap/endless/pkg/client/pd"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/matcher/metrics"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"github.com/spf13/cast"
	"github.com/tidwall/gjson"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
)

var regionSplitSizes = []string{"128MiB", "256MiB", "512MiB", "1024MiB", "5120MiB", "10240MiB"}
var dynamicRegionConfigs1 = []string{
	"coprocessor.enable-region-bucket",
	"coprocessor.region-bucket-size",
	"coprocessor.region-max-size",
	"coprocessor.region-max-keys",
	"coprocessor.region-split-size",
	"coprocessor.region-split-keys",
	"schedule.max-merge-region-size",
	"schedule.max-merge-region-keys",
}
var dynamicRegionConfigs2 = []string{
	"coprocessor.enable-region-bucket",
	"coprocessor.region-bucket-size",
	"coprocessor.region-split-size",
	"coprocessor.region-split-keys",
	"coprocessor.region-max-size",
	"coprocessor.region-max-keys",
	"schedule.max-merge-region-size",
	"schedule.max-merge-region-keys",
}

// keys = size / MiB * 10000
// region-max-size = region-split-size * 1.5
// max-merge-region-size = region-max-size * 13%  or max-merge-region-size = region-split-size * 20%
func getTargetDynamicRegionConfigsValue(regionSplitSize string) (map[string]string, error) {
	configsValue := make(map[string]string)
	sizeStr := regionSplitSize[0 : len(regionSplitSize)-3]
	unit := regionSplitSize[len(regionSplitSize)-3:]
	sizeValue, _ := strconv.ParseFloat(sizeStr, 64)
	if unit != "MiB" && unit != "GiB" {
		return configsValue, errors.New("unsupported region split size")
	}
	// convert it to MiB
	if unit == "GiB" {
		sizeValue = sizeValue * 1024
	}
	configsValue["coprocessor.enable-region-bucket"] = "true"
	configsValue["coprocessor.region-bucket-size"] = "96MiB"
	configsValue["coprocessor.region-split-size"] = regionSplitSize
	configsValue["coprocessor.region-split-keys"] = strconv.Itoa(int(math.Ceil(sizeValue * 10000)))
	configsValue["coprocessor.region-max-size"] = strconv.Itoa(int(math.Ceil(sizeValue*1.5))) + "MiB"
	configsValue["coprocessor.region-max-keys"] = strconv.Itoa(int(math.Ceil(sizeValue * 1.5 * 10000)))
	configsValue["schedule.max-merge-region-size"] = strconv.Itoa(int(math.Ceil(sizeValue * 0.2)))
	configsValue["schedule.max-merge-region-keys"] = strconv.Itoa(int(math.Ceil(sizeValue * 0.2 * 10000)))

	return configsValue, nil
}

func getCurrentRegionConfigsValue(tc resource.TiDBCluster) {
	log.Info("Get all the current region related configuration values.")
	for _, dynamicRegionConfig := range dynamicRegionConfigs1 {
		if strings.HasPrefix(dynamicRegionConfig, "coprocessor") {
			oltppkg.GetConfigByName(tc, "tikv", dynamicRegionConfig)
			if dynamicRegionConfig == "coprocessor.region-split-size" || dynamicRegionConfig == "coprocessor.region-max-size" {
				getPdSchedulerConfigByMetric(tc, strings.TrimPrefix(dynamicRegionConfig, "coprocessor."))
			}
		} else if strings.HasPrefix(dynamicRegionConfig, "schedule") {
			oltppkg.GetConfigByName(tc, "pd", dynamicRegionConfig)
			getPdSchedulerConfigByMetric(tc, strings.TrimPrefix(dynamicRegionConfig, "schedule."))
		}
	}
	oltppkg.ExecPdCtl(tc, "scheduler show")
}

func setRegionConfigOnline(ctx context.Context, tc resource.TiDBCluster, targetConfigsValue map[string]string) {
	select {
	case <-ctx.Done():
		log.Info("ctx done.")
		return
	default:
	}
	fmt.Printf("targe config value:\n%v\n", targetConfigsValue)
	var dynamicRegionConfigs []string
	isIntegral := func(val float64) bool {
		return val == float64(int(val))
	}
	regionSplitKeys, _ := strconv.Atoi(targetConfigsValue["coprocessor.region-split-keys"])
	currentRegionMaxKeys, _ := strconv.Atoi(oltppkg.GetConfigByName(tc, "tikv", "coprocessor.region-max-keys")[0].Value)
	if regionSplitKeys > currentRegionMaxKeys {
		dynamicRegionConfigs = dynamicRegionConfigs1
	} else {
		dynamicRegionConfigs = dynamicRegionConfigs2
	}
	log.Info("start update region config.")
	for _, dynamicRegionConfig := range dynamicRegionConfigs {
		targetValue := targetConfigsValue[dynamicRegionConfig]
		if strings.HasPrefix(dynamicRegionConfig, "coprocessor") {
			if dynamicRegionConfig == "coprocessor.region-max-size" || dynamicRegionConfig == "coprocessor.region-split-size" {
				convertToGiBValue := cast.ToFloat64(strings.TrimSuffix(targetValue, "MiB")) / 1024
				if isIntegral(convertToGiBValue) {
					targetValue = cast.ToString(convertToGiBValue) + "GiB"
				}
			}
			oltppkg.SetAndVerifyConfigOnline(tc, "tikv", dynamicRegionConfig, targetValue)
		} else if strings.HasPrefix(dynamicRegionConfig, "schedule") {
			setPdSchedulerConfig(tc, strings.TrimPrefix(dynamicRegionConfig, "schedule."), targetValue)
		}
	}
	addPdScheduler(tc, "split-bucket-scheduler")
	convertToValueOfMiB := func(val string) int {
		var mibValue int
		if strings.HasSuffix(val, "MiB") {
			mibValue = cast.ToInt(strings.TrimSuffix(val, "MiB"))
		} else if strings.HasSuffix(val, "GiB") {
			mibValue = cast.ToInt(strings.TrimSuffix(val, "GiB")) * 1024
		}
		return mibValue
	}
	targetRegionMaxSize := convertToValueOfMiB(targetConfigsValue["coprocessor.region-max-size"])
	targetRegionSplitSize := convertToValueOfMiB(targetConfigsValue["coprocessor.region-split-size"])
	Eventually(func() int {
		return cast.ToInt(getPdSchedulerConfigByMetric(tc, "region-max-size"))
	}, 300*time.Second, 10*time.Second).Should(Equal(targetRegionMaxSize))

	Eventually(func() int {
		return cast.ToInt(getPdSchedulerConfigByMetric(tc, "region-split-size"))
	}, 300*time.Second, 10*time.Second).Should(Equal(targetRegionSplitSize))
	log.Info("update region config success.")
}

func addPdScheduler(tc resource.TiDBCluster, scheduler string) {
	schedulers := Try(oltppkg.ExecPdCtl(tc, "scheduler show")).(string)
	if strings.Contains(schedulers, scheduler) {
		log.Info(fmt.Sprintf("%s has been added", scheduler))
	} else {
		r := Try(oltppkg.ExecPdCtl(tc, fmt.Sprintf("scheduler add %s", scheduler))).(string)
		Expect(strings.Contains(strings.ToLower(r), "success")).To(BeTrue())
		schedulers = Try(oltppkg.ExecPdCtl(tc, "scheduler show")).(string)
		Expect(strings.Contains(strings.ToLower(schedulers), scheduler)).To(BeTrue())
	}
}

func setPdSchedulerConfig(tc resource.TiDBCluster, config string, value string) {
	r := Try(oltppkg.ExecPdCtl(tc, fmt.Sprintf("config set %s %s ", config, value))).(string)
	Expect(strings.Contains(strings.ToLower(r), "success")).To(BeTrue())
	currentValue := getPdSchedulerConfigByPdCtl(tc, config)
	Expect(cast.ToInt(value)).Should(Equal(cast.ToInt(currentValue)))
}

func getPdSchedulerConfigByPdCtl(tc resource.TiDBCluster, config string) float64 {
	bytes := []byte(Try(oltppkg.ExecPdCtl(tc, "config show")).(string))
	value := gjson.GetBytes(bytes, fmt.Sprintf("schedule.%s", config)).Value().(float64)
	log.Info("pd scheduler config from pd ctl", zap.String("config", config), zap.Float64("value", value))
	return value
}

func getPdSchedulerConfigByMetric(tc resource.TiDBCluster, config string) float64 {
	promURL := Try(tc.ServiceURL(resource.Prometheus)).(*url.URL)
	query := fmt.Sprintf("pd_config_status{type=\"%s\"}", config)
	value := Try(oltppkg.GetMetric(promURL, query, time.Now())).(float64)
	log.Info("pd scheduler config from metric", zap.String("config", config), zap.Float64("value", value))
	return value
}

func updateRegionConfigOnlineRepeatedly(
	ctx context.Context,
	tc resource.TiDBCluster,
	g *errgroup.Group,
	regionSplitSizes []string,
	runTime time.Duration,
	runTimePerConfig time.Duration,
	pr *io.PipeReader,
	qpsJitterCheckParams oltppkg.QpsJitterCheckParameters,
) {
	regionSizeCheckParms := oltppkg.RegionSizeCheckParameters{
		IsCheck: false,
	}
	oltppkg.FunCommonCheck(ctx, tc, g, runTime, pr, qpsJitterCheckParams, regionSizeCheckParms)
	g.Go(WithGinkgoRecover(func() {
		for {
			select {
			case <-ctx.Done():
				log.Info("ctx done.")
				return
			default:
			}
			for _, regionSplitSize := range regionSplitSizes {
				oltppkg.WaitFor(ctx, "run with current config for", runTimePerConfig)
				regionConfig := Try(getTargetDynamicRegionConfigsValue(regionSplitSize)).(map[string]string)
				setRegionConfigOnline(ctx, tc, regionConfig)
			}
		}
	}))
}

func rollingUpdateRegionConfig(ctx context.Context, tc resource.TiDBCluster, targeConfigsValue map[string]string, timeout time.Duration) {
	select {
	case <-ctx.Done():
		log.Info("ctx done.")
		return
	default:
	}
	currentRegionSplitMaxSize := oltppkg.GetConfigByName(tc, "tikv", "coprocessor.region-split-size")[0].Value
	currentEnableRegionBucket := oltppkg.GetConfigByName(tc, "tikv", "coprocessor.enable-region-bucket")[0].Value
	targeRegionSplitSize := targeConfigsValue["coprocessor.region-split-size"]
	targeEnableRegionBucket := targeConfigsValue["coprocessor.enable-region-bucket"]
	if currentRegionSplitMaxSize != targeRegionSplitSize || currentEnableRegionBucket != targeEnableRegionBucket {
		fmt.Printf("targe config value:\n%v\n", targeConfigsValue)
		tikvConfig := fmt.Sprintf("[coprocessor]\nenable-region-bucket = %s\nregion-split-size = \"%s\"",
			targeEnableRegionBucket, targeRegionSplitSize)
		fmt.Println(tikvConfig)
		oltppkg.RollingUpdateConfig(tc, resource.TiKV, tikvConfig, timeout)
		setPdSchedulerConfig(tc, "max-merge-region-size", targeConfigsValue["schedule.max-merge-region-size"])
		setPdSchedulerConfig(tc, "max-merge-region-keys", targeConfigsValue["schedule.max-merge-region-keys"])
		addPdScheduler(tc, "split-bucket-scheduler")
		for _, dynamicRegionConfig := range dynamicRegionConfigs1 {
			if !strings.HasPrefix(dynamicRegionConfig, "schedule") {
				configRows := oltppkg.GetConfigByName(tc, "tikv", dynamicRegionConfig)
				for _, configRow := range configRows {
					currentValue := configRow.Value
					if strings.HasSuffix(currentValue, "GiB") {
						currentValue = cast.ToString(cast.ToInt(cast.ToInt(strings.TrimSuffix(currentValue, "GiB"))*1024)) + "MiB"
					}
					Expect(currentValue).To(Equal(targeConfigsValue[dynamicRegionConfig]))
				}
				if dynamicRegionConfig == "coprocessor.region-split-size" || dynamicRegionConfig == "coprocessor.region-max-size" {
					getPdSchedulerConfigByMetric(tc, strings.TrimPrefix(dynamicRegionConfig, "coprocessor."))
				}
			}
		}
		log.Info("update region config success.")
	}
}

func rollingUpdateRegionConfigRepeatedly(
	ctx context.Context,
	tc resource.TiDBCluster,
	g *errgroup.Group,
	regionSplitSizes []string,
	runTime time.Duration,
	runTimePerConfig time.Duration,
	pr *io.PipeReader,
	qpsJitterCheckParams oltppkg.QpsJitterCheckParameters,
) {
	regionSizeCheckParms := oltppkg.RegionSizeCheckParameters{
		IsCheck: false,
	}
	for {
		select {
		case <-ctx.Done():
			log.Info("ctx done.")
			return
		default:
		}
		for _, regionSplitSize := range regionSplitSizes {
			regionConfig := Try(getTargetDynamicRegionConfigsValue(regionSplitSize)).(map[string]string)
			oltppkg.FunCommonCheck(ctx, tc, g, runTimePerConfig, pr, qpsJitterCheckParams, regionSizeCheckParms)
			oltppkg.WaitFor(ctx, "run with current config for", runTimePerConfig)
			updateConfigTimeout := 60 * time.Minute
			g.Go(WithGinkgoRecover(func() {
				rollingUpdateRegionConfig(ctx, tc, regionConfig, updateConfigTimeout)
			}))
			ConsistentlyWithMetrics(tc, updateConfigTimeout).
				ShouldNot(QPSDropMoreThanOffset(0.5, "2m").For(30 * time.Second).WithCtx(ctx))
		}
	}
}

func regionsCheckAfterCaseEnd(ctx context.Context, tc resource.TiDBCluster, benchToolset resource.WorkloadNode) {
	oltppkg.StopWorkload(benchToolset, "goTpc")
	oltppkg.StopWorkload(benchToolset, "sysBench")
	oltppkg.LogAndBy("Waiting for oversized regions and undersized regions to be less than a certain value after workload stopping.")
	client := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	Try(client.SetConfig("split-merge-interval", "10m"))
	Try(oltppkg.ExecPdCtl(tc, "config set merge-schedule-limit 200"))
	defer func() {
		Try(oltppkg.ExecPdCtl(tc, "config set merge-schedule-limit 8"))
		Try(client.SetConfig("split-merge-interval", "30m"))
	}()
	oltppkg.WaitForOversizedRegionsLessThan(ctx, tc, 120*time.Minute, 2*time.Minute, 1)
	oltppkg.WaitForUndersizedRegionsLessThan(ctx, tc, 120*time.Minute, 2*time.Minute, 1)
}