package configupdate

import (
	"context"
	"errors"
	"io"
	"math"
	"regexp"
	"strconv"
	"strings"
	"time"

	. "github.com/onsi/gomega"
	"github.com/spf13/cast"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"

	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
)

func OLTP_CONFIG_UPDATE_CONFIG_FUN_001(
	ctx context.Context,
	tc resource.TiDBCluster,
	pr *io.PipeReader,
	caseCmdParms oltppkg.CaseCmdParms,
) {
	oltppkg.LogAndBy("Modify configuration online")
	tikvVer := oltppkg.GetCompVersion(ctx, tc, "tikv")
	curVerShort := strings.TrimSuffix(strings.TrimPrefix(tikvVer, "v"), "-pre")
	if oltppkg.CompareVersion(curVerShort, "5.4.0") < 0 {
		log.Info("Skip this case for current version is lower than 5.4.0", zap.String("tikvVersion", curVerShort))
		return
	}
	tikvCPUNum := caseCmdParms.ClusterSpecParms.TiKVCPUNum

	type configInfo struct {
		supportVersion string
		minValue       string
		maxValue       string
		initValue      string
		targetValue    string
	}

	strToInt := func(v string) int {
		r, err := strconv.Atoi(v)
		Expect(err).Should(Succeed())
		return r
	}

	intToStr := func(v int) string {
		r := strconv.Itoa(v)
		return r
	}

	isCapacity := func(v string) bool {
		if strings.HasSuffix(v, "KiB") || strings.HasSuffix(v, "MiB") || strings.HasSuffix(v, "GiB") || strings.HasSuffix(v, "TiB") {
			return true
		}
		return false
	}

	isNumber := func(v string) bool {
		match, _ := regexp.MatchString("([a-z]+)", v)
		return !match
	}

	isDuration := func(v string) bool {
		var durationUnits = []string{"ms", "s", "h", "d"}
		for _, durationUnit := range durationUnits {
			if strings.HasSuffix(v, durationUnit) {
				return true
			}
		}
		return false
	}

	parseCapacity := func(v string) (string, string) {
		vStr := v[0 : len(v)-3]
		unit := v[len(v)-3:]
		return vStr, unit
	}

	parseDuration := func(v string) (string, string) {
		var durationUnits = []string{"ms", "s", "h", "d"}
		for _, durationUnit := range durationUnits {
			if strings.HasSuffix(v, durationUnit) {
				return strings.TrimSuffix(v, durationUnit), durationUnit
			}
		}
		return "", ""
	}

	validCheck := func(v string, min string, max string) {
		if strToInt(v) > strToInt(max) || strToInt(v) < strToInt(min) {
			err := errors.New("invalid config")
			Expect(err).Should(Succeed())
		}
	}

	getTargetValue := func(conf string, configInfo configInfo) string {
		var targetValueInt int
		var targetValue string
		var initValue string = configInfo.initValue
		var minValue string = configInfo.minValue
		var maxValue string = configInfo.maxValue
		if isNumber(initValue) {
			log.Info("config is number")
			validCheck(initValue, minValue, maxValue)
			if initValue == maxValue {
				targetValueInt = strToInt(initValue) - 1
			} else {
				targetValueInt = strToInt(initValue) + 1
			}
			targetValue = intToStr(targetValueInt)
		} else if isCapacity(initValue) {
			log.Info("config is capacity")
			initValueStr, unit := parseCapacity(initValue)
			targetValueInt = strToInt(initValueStr) + 1
			targetValue = intToStr(targetValueInt) + unit
		} else if isDuration(initValue) {
			log.Info("config is duration")
			initValueStr, unit := parseDuration(initValue)
			targetValueInt = strToInt(initValueStr) + 1
			targetValue = intToStr(targetValueInt) + unit
		}
		return targetValue
	}

	configsInfo := map[string]*configInfo{
		"raftstore.apply-pool-size":      &configInfo{supportVersion: "5.4.0", minValue: "1", maxValue: intToStr(10 * tikvCPUNum), initValue: ""},
		"raftstore.store-pool-size":      &configInfo{supportVersion: "5.4.0", minValue: "1", maxValue: intToStr(10 * tikvCPUNum), initValue: ""},
		"raftdb.max-background-flushes":  &configInfo{supportVersion: "5.4.0", minValue: "1", maxValue: intToStr(10 * tikvCPUNum), initValue: ""},
		"rocksdb.max-background-flushes": &configInfo{supportVersion: "5.4.0", minValue: "1", maxValue: intToStr(10 * tikvCPUNum), initValue: ""},
		"raftdb.max-background-jobs":     &configInfo{supportVersion: "5.4.0", minValue: "1", maxValue: intToStr(10 * tikvCPUNum), initValue: ""},
		"rocksdb.max-background-jobs":    &configInfo{supportVersion: "5.4.0", minValue: "1", maxValue: intToStr(10 * tikvCPUNum), initValue: ""},

		"readpool.unified.max-thread-count":  &configInfo{supportVersion: "6.0.0", minValue: "1", maxValue: intToStr(int(math.Max(4, float64(tikvCPUNum)))), initValue: ""},
		"backup.num-threads":                 &configInfo{supportVersion: "6.0.0", minValue: "1", maxValue: intToStr(tikvCPUNum), initValue: ""},
		"storage.scheduler-worker-pool-size": &configInfo{supportVersion: "6.0.0", minValue: "1", maxValue: intToStr(tikvCPUNum), initValue: ""},
		"raftstore.raft-max-inflight-msgs":   &configInfo{supportVersion: "6.0.0", minValue: "1", maxValue: "16384", initValue: ""},
		"raftstore.apply-max-batch-size":     &configInfo{supportVersion: "6.0.0", minValue: "1", maxValue: "10240", initValue: ""},
		"raftstore.store-max-batch-size":     &configInfo{supportVersion: "6.0.0", minValue: "1", maxValue: "10240", initValue: ""},
		"raftstore.raft-max-size-per-msg":    &configInfo{supportVersion: "6.0.0", minValue: "1KiB", maxValue: "3GiB", initValue: ""}, // [1B,3GiB]

		"quota.foreground-cpu-time":        &configInfo{supportVersion: "6.1.0", minValue: "0", maxValue: "9223372036854775807", initValue: "", targetValue: "8000"},  // [0, 9223372036854775807]
		"quota.foreground-write-bandwidth": &configInfo{supportVersion: "6.1.0", minValue: "0KiB", maxValue: "8387584PiB", initValue: "", targetValue: "1000MiB"},     // [0, 8192PB]
		"quota.foreground-read-bandwidth":  &configInfo{supportVersion: "6.1.0", minValue: "0KiB", maxValue: "8387584PiB", initValue: "", targetValue: "1000MiB"},     // [0, 8192PB]
		"quota.max-delay-duration":         &configInfo{supportVersion: "6.1.0", minValue: "0", maxValue: "213503d", initValue: "", targetValue: "500ms"},             // [0, 213503d]
		"server.grpc-memory-pool-quota":    &configInfo{supportVersion: "6.1.0", minValue: "1", maxValue: "9223372036854775807", initValue: "", targetValue: "10PiB"}, // (0, 8192PB], deafult: 9223372036854776000
		"server.max-grpc-send-msg-len":     &configInfo{supportVersion: "6.1.0", minValue: "1", maxValue: "2147483647", initValue: ""},                                // (0, 2147483647)
		"server.raft-msg-max-batch-size":   &configInfo{supportVersion: "6.1.0", minValue: "1", maxValue: "9223372036854775807", initValue: ""},                       // (0, 9223372036854775807)
		"raftstore.raft-entry-max-size":    &configInfo{supportVersion: "6.1.0", minValue: "1KiB", maxValue: "3GiB", initValue: ""},                                   // (0, 3GiB]
	}

	getInitConfig := func(configsInfo map[string]*configInfo) {
		oltppkg.LogAndBy("Get initial config")
		for conf, _ := range configsInfo {
			if oltppkg.CompareVersion(curVerShort, configsInfo[conf].supportVersion) < 0 {
				continue
			}
			initValue := (oltppkg.GetConfigByName(tc, "tikv", conf))[0].Value
			configsInfo[conf].initValue = initValue
		}
	}

	recoverConfig := func(configsInfo map[string]*configInfo, curVerShort string) {
		oltppkg.LogAndBy("Recover to initial config")
		for conf, _ := range configsInfo {
			if oltppkg.CompareVersion(curVerShort, configsInfo[conf].supportVersion) < 0 {
				continue
			}
			initValue := configsInfo[conf].initValue
			targetValue := initValue
			if conf == "server.grpc-memory-pool-quota" && cast.ToUint64(initValue) > cast.ToUint64(configsInfo[conf].maxValue) {
				targetValue = "8192PiB"
			}
			log.Info("config-recover", zap.String("config", conf), zap.String("initValue", initValue), zap.String("targetValue", targetValue))
			oltppkg.SetConfigOnline(tc, "tikv", conf, initValue)
			configRows := oltppkg.GetConfigByName(tc, "tikv", conf)
			for _, configRow := range configRows {
				currentValue := configRow.Value
				Expect(currentValue).To(Equal(targetValue))
			}
		}
	}
	configNum := len(configsInfo)
	modifyInterval := 2 * time.Minute
	warmUpTime := 10 * time.Minute
	recoveryTime := 10 * time.Minute
	runTime := time.Duration(int(modifyInterval)*configNum) + warmUpTime + recoveryTime + 20*time.Minute
	log.Info("Test will run ", zap.Duration("TotalRunTime", runTime))
	ctxAll, cancel := context.WithTimeout(ctx, runTime)
	defer cancel()
	g, ctxg := errgroup.WithContext(ctxAll)

	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 0.5,
		TolerateDur:  30 * time.Second,
	}

	oltppkg.FunCommonCheck(ctxg, tc, g, runTime, pr, qpsJitterCheckParms)

	g.Go(WithGinkgoRecover(func() {
		oltppkg.WaitFor(ctxg, "warm up before modify configuration", warmUpTime)
		getInitConfig(configsInfo)
		defer recoverConfig(configsInfo, curVerShort)
		oltppkg.LogAndBy("Modify config online")
		for conf, _ := range configsInfo {
			if oltppkg.CompareVersion(curVerShort, configsInfo[conf].supportVersion) < 0 {
				log.Info("this config do not support modfiy config online in current version", zap.String("tikvCurVersion", curVerShort), zap.String("configName", conf))
				continue
			}
			var targetValue string
			var initValue string = configsInfo[conf].initValue
			var minValue string = configsInfo[conf].minValue
			var maxValue string = configsInfo[conf].maxValue
			var preSetTargetValue string = configsInfo[conf].targetValue
			log.Info("config-show", zap.String("configName", conf), zap.String("initValue", initValue), zap.String("minValue", minValue), zap.String("maxValue", maxValue))
			if conf == "server.grpc-memory-pool-quota" {
				targetValue = preSetTargetValue
			} else if conf == "quota.foreground-cpu-time" || conf == "quota.foreground-write-bandwidth" || conf == "quota.foreground-read-bandwidth" || conf == "quota.max-delay-duration" {
				if initValue == "0" || initValue == "0s" || initValue == "0KiB" {
					Expect(preSetTargetValue).NotTo(Equal(""))
					targetValue = preSetTargetValue
				} else {
					targetValue = getTargetValue(conf, *configsInfo[conf])
				}
			} else {
				targetValue = getTargetValue(conf, *configsInfo[conf])
			}
			oltppkg.SetAndVerifyConfigOnline(tc, "tikv", conf, targetValue)
			oltppkg.WaitFor(ctxg, "wait for next config modifing", modifyInterval)
		}

		oltppkg.WaitFor(ctxg, "wait before recover config", recoveryTime)
		recoverConfig(configsInfo, curVerShort)
	}))

	Expect(g.Wait()).Should(BeNil())
}