package oltppkg

import (
	"fmt"
	"math"
	"net/url"
	"regexp"
	"strconv"
	"strings"
	"time"

	. "github.com/onsi/gomega"
	"github.com/onsi/gomega/types"
	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/workloadnode/cluster"
	"github.com/pingcap/endless/pkg/workloadnode/tiup"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
	"github.com/tidwall/gjson"
	"go.uber.org/zap"
)

const (
	memberhealthPrefix = "/pd/api/v1/health"
	pdLeaderPrefix     = "/pd/api/v1/leader"
	storePrefix        = "/pd/api/v1/stores"
	pdMemberfix        = "/pd/api/v1/members"
)

func SetStoreLimit(tc resource.TiDBCluster, limit int) {
	log.Info("Set store limit to ", zap.Int("All-stores-limit", limit))
	pd := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	Try(pd.SetAllStoreLimit(60))
}
func TryGetPdLeaderName(tc resource.TiDBCluster) string {
	pd, err := pdclient.NewPDClient(tc)
	if err != nil {
		return ""
	}
	resp, err := pd.Get(pdLeaderPrefix)
	if err != nil {
		return ""
	}
	leader, ok := JSONSelector(resp, "name").(string)
	if ok {
		log.Info("Get leader name", zap.String("leader", leader))
		return leader
	} else {
		return ""
	}
}
func GetPdLeaderName(tc resource.MySQLCompatible) string {
	resp := Try(BasicClusterExecPdCtl(tc, "member leader show")).(string)
	return gjson.GetBytes([]byte(resp), "name").Value().(string)
}

func GetPdMembersName(tc resource.MySQLCompatible) []string {
	bytes := []byte(Try(BasicClusterExecPdCtl(tc, "member")).(string))
	result := gjson.GetBytes(bytes, "members.#.name").Value()
	var pdMembers []string
	for _, pdMember := range result.([]interface{}) {
		pdMembers = append(pdMembers, pdMember.(string))
	}
	return pdMembers
}

func CheckPdMemberUp(tc resource.MySQLCompatible) {
	bytes := []byte(Try(BasicClusterExecPdCtl(tc, "health")).(string))
	result := gjson.GetBytes(bytes, "#.health").Value()
	Expect(result).Should(All(func() types.GomegaMatcher { return BeTrue() }))
}
func CheckTiKVUp(tc resource.MySQLCompatible) {
	bytes := []byte(Try(BasicClusterExecPdCtl(tc, "store")).(string))
	result := gjson.GetBytes(bytes, "stores.#.store.state_name").Value()
	Expect(result).Should(All(func() types.GomegaMatcher { return Equal("Up") }))
}
func CheckClusterHealth(tc resource.MySQLCompatible) {
	CheckTiKVUp(tc)
	CheckPdMemberUp(tc)
	CheckRegionHealth(tc)
	Expect(GetPdLeaderName(tc) != "").To(BeTrue())
}

func SetPDRandomScheduler(tc resource.MySQLCompatible, command string) {
	var cmdlist []string
	cmdlist = append(cmdlist, fmt.Sprintf("schedule %s shuffle-leader-scheduler", command))
	cmdlist = append(cmdlist, fmt.Sprintf("schedule %s shuffle-region-scheduler", command))
	cmdlist = append(cmdlist, fmt.Sprintf("schedule %s random-merge-scheduler", command))

	for i := 0; i < len(cmdlist); i++ {
		Try(BasicClusterExecPdCtl(tc, cmdlist[i]))
	}
}
func SetPDScheduler(tc resource.TiDBCluster, command string, pauseSecond int) {
	var cmdlist []string
	if command == "pause" {
		cmdlist = append(cmdlist, fmt.Sprintf("schedule %s balance-hot-region-scheduler %d", command, pauseSecond))
		cmdlist = append(cmdlist, fmt.Sprintf("schedule %s balance-leader-scheduler %d", command, pauseSecond))
		cmdlist = append(cmdlist, fmt.Sprintf("schedule %s balance-region-scheduler %d", command, pauseSecond))
	} else {
		cmdlist = append(cmdlist, fmt.Sprintf("schedule %s balance-hot-region-scheduler", command))
		cmdlist = append(cmdlist, fmt.Sprintf("schedule %s balance-leader-scheduler", command))
		cmdlist = append(cmdlist, fmt.Sprintf("schedule %s balance-region-scheduler", command))
	}

	for i := 0; i < len(cmdlist); i++ {
		Try(ExecPdCtl(tc, cmdlist[i]))
	}
}
func ExecPdCtl(tc resource.TiDBCluster, cmd string) (string, error) {
	pdcmd := fmt.Sprintf("/pd-ctl %s", cmd)
	stdout, stderr, exitCode, err := tc.Exec(resource.TiDBClusterExecOptions{Target: resource.PD, Command: pdcmd})
	if stdout != "" {
		//fmt.Printf("%s\n%v\n", pdcmd, stdout)
		return stdout, nil
	}
	if err != nil {
		log.Info(fmt.Sprintf("stdout:%s\n stderr:%s\n exitCode:%d\n", stdout, stderr, exitCode))
		return "", err
	}
	return "", nil
}
func CheckRegionHealth(tc resource.MySQLCompatible) bool {
	for _, v := range []string{"miss-peer", "down-peer", "offline-peer"} {
		bytes := []byte(Try(BasicClusterExecPdCtl(tc, "region check "+v)).(string))
		t := gjson.GetBytes(bytes, "count").Value()
		if t.(float64) != float64(0) {
			log.Info("Check fail, region is not 0.", zap.String("region type", v), zap.Float64("region count", t.(float64)))
			return false
		}
		log.Info("Check pass, region is 0.", zap.String("region type", v))
	}
	return true
}

func WaitForTiKVUP(tc resource.MySQLCompatible, timeOut time.Duration) {
	Eventually(func() interface{} {
		bytes := []byte(Try(BasicClusterExecPdCtl(tc, "store")).(string))
		result := gjson.GetBytes(bytes, "stores.#.store.state_name").Value()
		return result
	}, timeOut).Should(All(func() types.GomegaMatcher { return Equal("Up") }))
}

func WaitForTikvCount(tc resource.MySQLCompatible, targetTikvCount int, timeOut time.Duration) {
	Eventually(func() int {
		bytes := []byte(Try(BasicClusterExecPdCtl(tc, "store")).(string))
		count := gjson.GetBytes(bytes, "count").Float()
		return int(count)
	}, timeOut, 1*time.Minute).Should(Equal(targetTikvCount))
}

func WaitForRegionHealth(tc resource.MySQLCompatible, timeOut time.Duration) {
	Eventually(func() bool {
		return CheckRegionHealth(tc)
	}, timeOut, 10*time.Second).Should(BeTrue())
}

// GetStoreSizeByType retrieve store available/used_size/capacity size and convert it to MiB
func GetStoreSizeByType(tc resource.MySQLCompatible, sizeType string) ([]float64, float64, error) {
	bytes := []byte(Try(BasicClusterExecPdCtl(tc, "store")).(string))
	result := gjson.GetBytes(bytes, fmt.Sprintf("stores.#.status.%s", sizeType)).Array()

	storeSize := []float64{}
	sumSize := float64(0)
	for _, s := range result {
		s := s.String()
		reg := regexp.MustCompile(`(\d+\.*\d*)(\S+)`)
		s1 := reg.FindAllStringSubmatch(s, -1)
		if len(s1) != 1 {
			return nil, 0, fmt.Errorf("not match reg")
		}
		if len(s1[0]) != 3 {
			return nil, 0, fmt.Errorf("not match reg")
		}
		capacity, err := strconv.ParseFloat(s1[0][1], 64)
		if err != nil {
			return nil, sumSize, err
		}

		unit := s1[0][2]
		if unit == "GiB" {
			capacity = capacity * 1024
		} else if unit == "TiB" {
			capacity = capacity * 1024 * 1024
		} else if unit == "B" {
			capacity = 0
		} else if unit == "KiB" {
			capacity = capacity / 1024
		} else if unit == "MiB" {
			capacity = capacity * 1
		} else {
			return nil, 0, fmt.Errorf("not match reg")
		}
		sumSize = sumSize + capacity
		storeSize = append(storeSize, capacity)
	}
	return storeSize, sumSize, nil
}

// GetNodeStoreSizeByType
// nodeType can be = ["tikv", "tiflash"]
func GetNodeStoreSizeByType(tc resource.TiDBCluster, nodeType string, sizeType string) (storeSizes []float64, err error) {
	resp := Try(BasicClusterExecPdCtl(tc, "store")).(string)
	// print(string(bytes))
	nodeTypeResult := gjson.GetBytes([]byte(resp), fmt.Sprintf("stores.#.store.address")).Array()
	storeSizeResult := gjson.GetBytes([]byte(resp), fmt.Sprintf("stores.#.status.%s", sizeType)).Array()
	for i, v := range nodeTypeResult {
		if strings.Contains(v.String(), nodeType) {
			store := storeSizeResult[i].String()
			reg := regexp.MustCompile(`(\d+\.*\d*)(\S+)`)
			// expect regexp result = [origin string, store size num, unit]
			regexpResult := reg.FindAllStringSubmatch(store, -1)
			if len(regexpResult) != 1 {
				return nil, fmt.Errorf("not match reg")
			}
			if len(regexpResult[0]) != 3 {
				return nil, fmt.Errorf("not match reg")
			}
			capacity, err := strconv.ParseFloat(regexpResult[0][1], 64)
			if err != nil {
				return nil, err
			}

			unit := regexpResult[0][2]
			if unit == "GiB" {
				capacity = capacity * 1024
			} else if unit == "TiB" {
				capacity = capacity * 1024 * 1024
			} else if unit == "B" {
				capacity = 0
			} else if unit == "KiB" {
				capacity = capacity / 1024
			} else if unit == "MiB" {
				capacity = capacity * 1
			} else {
				return nil, fmt.Errorf("not match reg")
			}
			storeSizes = append(storeSizes, capacity)
		}
	}
	return storeSizes, nil
}

// CheckIfHeterogeneousDisk check if diviation of all store capacity larger than 0.2
func CheckIfHeterogeneousDisk(tc resource.MySQLCompatible) bool {
	storeCapacityList, _, err := GetStoreSizeByType(tc, "capacity")
	Expect(err).Should(BeNil())
	min, max := FindMinAndMax(storeCapacityList)
	avg := (max + min) / float64(2)
	deviation := math.Max(math.Abs(avg-min)/avg, math.Abs(max-avg)/avg)
	log.Info("Deviation of all tikv capacity", zap.Float64("deviation", deviation))
	return deviation > 0.2
}

// GetStoreIdAddrMap
func GetStoreIdAddrMap(pdURL *url.URL, key string, value string) map[string]string {
	pd := Try(pdclient.NewPDClientVarUrl(pdURL)).(*pdclient.Client)
	bytes := Try(pd.GetAllStoresInfo()).([]byte)
	// print(string(bytes))
	resultMap := make(map[string]string)
	result := gjson.GetBytes(bytes, fmt.Sprintf("stores.#.store.%s", key)).Array()

	var idList []string
	for _, s := range result {
		id := s.String()
		idList = append(idList, id)
	}
	result = gjson.GetBytes(bytes, fmt.Sprintf("stores.#.store.%s", value)).Array()

	var valueList []string
	for _, s := range result {
		id := strings.Split(s.String(), ".")[0]
		valueList = append(valueList, id)
	}
	for idx, s := range idList {
		resultMap[s] = valueList[idx]
	}

	return resultMap
}

// GetReplicationStatus
func GetRepliactionStatus(tc resource.MySQLCompatible) string {
	pdURL := Try(tc.ServiceURL(resource.PDAddr)).(*url.URL)
	pd := Try(pdclient.NewPDClientVarUrl(pdURL)).(*pdclient.Client)
	bytes := Try(pd.GetJsonBytes("/pd/api/v1/replication_mode/status")).([]byte)
	// print(string(bytes))
	result := gjson.GetBytes(bytes, "dr-auto-sync.state").String()
	log.Info("Current replica status", zap.String("dr-auto-sync.state", result))
	return result
}

func TiupExecPdCtl(
	tc *tiup.TiUPCluster,
	pdCtlCmd string,
) (string, error) {
	clusterMeta := tc.ClusterInfo.ClusterMeta
	version := clusterMeta.ClusterVersion
	tiupNode := tc.ControlNodeHost
	pdURL := Try(tc.ServiceURL(resource.PDAddr)).(*url.URL)
	cmd := fmt.Sprintf("tiup ctl:%s pd -u %s %s", version, pdURL.String(), pdCtlCmd)
	if clusterMeta.TLSEnabled {
		pdHost := strings.Split(regexp.MustCompile("https?://").ReplaceAllString(pdURL.Host, ""), ".")[0]
		cmd = fmt.Sprintf("tiup ctl:%s pd -u https://%s:%s %s ", version, pdHost, pdURL.Port(), pdCtlCmd) +
			fmt.Sprintf("--cacert %s --cert %s --key %s", clusterMeta.TLSCACert, clusterMeta.TLSClientCert, clusterMeta.TLSClientKey)
	}
	log.Info("run pd ctl command", zap.String("pdCmd", cmd))
	return tiupNode.Exec(cmd, 2*time.Minute)
}

func BasicClusterExecPdCtl(tc cluster.BaseCluster, command string) (string, error) {
	_, ok := tc.(resource.TiDBCluster)
	if ok {
		return ExecPdCtl(tc.(resource.TiDBCluster), command)
	} else {
		return TiupExecPdCtl(tc.(*tiup.TiUPCluster), command)
	}
}
func GetComponentMapByType(ctx core.TestContext, tc *tiup.TiUPCluster, role string, status string) (nodeList []string, workloadNodeList []resource.WorkloadNode) {
	for _, instance := range tc.ClusterInfo.Instances {
		if strings.EqualFold(instance.Role, role) && strings.EqualFold(instance.Status, status) {
			nodeName := regexp.MustCompile("(.*)-peer").FindStringSubmatch(instance.Host)[1]
			nodeList = append(nodeList, instance.Host)
			workloadNode := ctx.Resource(nodeName).(resource.WorkloadNode)
			workloadNodeList = append(workloadNodeList, workloadNode)
		}
	}
	return nodeList, workloadNodeList
}