package pdfun

import (
	"context"
	"fmt"
	"reflect"
	"regexp"
	"strconv"
	"strings"
	"time"

	. "github.com/onsi/gomega"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"github.com/tidwall/gjson"
	"go.uber.org/zap"
	k8swait "k8s.io/apimachinery/pkg/util/wait"
)

func GetIngoreErrors(runType string) []string {
	oltpFunIgnoreErrors := strings.Split("2013,1213,1105,1205,8022,8027,8028,9004,9007,1062", ",")
	//Error 9002: TiKV server timeout
	//9001: PD server timeout
	//8027: Information schema is out of date: schema failed to update in 1 lease
	oltpRelIgnoreErrors := append(oltpFunIgnoreErrors, strings.Split("9002,9001", ",")...)
	switch runType {
	case "fun":
		return oltpFunIgnoreErrors
	case "rel":
		return oltpRelIgnoreErrors
	default:
		return oltpFunIgnoreErrors
	}
}

func getMinMaxAvailableRatio(ctx context.Context, tc resource.TiDBCluster) (float64, float64) {
	storeUseSize, _, err := oltppkg.GetStoreSizeByType(tc, "used_size")
	Expect(err).To(BeNil())
	storesAvailableSize, _, err := oltppkg.GetStoreSizeByType(tc, "available")
	Expect(err).To(BeNil())
	var ratio []float64
	for id, useSize := range storeUseSize {
		ratio = append(ratio, storesAvailableSize[id]/(storesAvailableSize[id]+useSize))
	}
	log.Info(fmt.Sprintf("Get current store available size %v", storesAvailableSize))
	log.Info(fmt.Sprintf("Get current store available ratio %v", ratio))
	minRatio, maxRatio := oltppkg.FindMinAndMax(ratio)
	return minRatio, maxRatio
}
func getMinMaxAvailableSize(ctx context.Context, tc resource.TiDBCluster) (float64, float64) {
	storesAvailableSize, _, err := oltppkg.GetStoreSizeByType(tc, "available")
	Expect(err).To(BeNil())
	log.Info(fmt.Sprintf("Get current store available size %v", storesAvailableSize))
	minAvail, maxAvail := oltppkg.FindMinAndMax(storesAvailableSize)
	return minAvail, maxAvail
}
func getRunInsertFunc(ctx context.Context, tc resource.TiDBCluster, workloadRunParms oltppkg.WorkloadRunParms, intervalTime time.Duration) k8swait.ConditionWithContextFunc {
	insertFunc := func(ctx context.Context) (bool, error) {
		dbURL, _ := tc.ServiceURL(resource.DBAddr)
		timeStr := time.Now().Format("20060102150405")
		dbname := fmt.Sprintf("fordata%s", timeStr)
		Try(oltppkg.RecreateDatabase(ctx, tc, dbname))
		sysbenchcmd := oltppkg.NewSysbench("oltp_read_only", "prepare").Db(dbname).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(100).
			Tables(100).TableSize(1000 * 10000).CreateSecondary("off")
		_, err := oltppkg.ExecWorkloadCmd(ctx, workloadRunParms.WorkloadNode, sysbenchcmd.Build(), true, resource.NopCloser(workloadRunParms.Cmdpw))
		if err != nil {
			return true, err
		}
		minRatio, _ := getMinMaxAvailableRatio(ctx, tc)
		if minRatio < 0.10 {
			oltppkg.WaitFor(ctx, "waiting for balance", intervalTime)
		}
		return false, nil
	}
	return insertFunc
}
func getStopWorkloadFunc(ctx context.Context, tc resource.TiDBCluster, workloadRunParms oltppkg.WorkloadRunParms, stopRunningFlag chan bool) k8swait.ConditionWithContextFunc {
	stopWorkloadFunc := func(ctx context.Context) (bool, error) {
		log.Info("running stop workload check")
		select {
		case <-stopRunningFlag:
			oltppkg.StopWorkload(workloadRunParms.WorkloadNode, "sysBench")
			return true, nil
		default:
			return false, nil
		}
	}
	return stopWorkloadFunc
}
func getDiskFullCheckFunc(ctx context.Context, tc resource.TiDBCluster, stopRunningFlag chan bool) k8swait.ConditionWithContextFunc {
	stopWorkloadFunc := func(ctx context.Context) (bool, error) {
		log.Info("running if disk full check")
		minAvail, maxAvail := getMinMaxAvailableSize(ctx, tc)
		if minAvail < 5120 {
			log.Info("Now the min available size should less than 5GB", zap.Float64("MinSize", minAvail))
			Expect(maxAvail).Should(BeNumerically("<", float64(100*1024)))
			log.Info("Pd schedule when disk if full check succeeded.")
			log.Info("Send stopRunningFlag")
			stopRunningFlag <- true
			log.Info("Send stopRunningFlag finished")
			return true, nil
		}
		return false, nil
	}
	return stopWorkloadFunc
}

func GetLeaderBalanceTime(tc resource.TiDBCluster) time.Duration {
	_, sumSize, err := oltppkg.GetStoreSizeByType(tc, "used_size")
	Expect(err).Should(BeNil())
	//check if > 2T
	if sumSize > 2*1024*1024 {
		return 6 * time.Minute
	} else if sumSize > 1*1024*1024 {
		return 3 * time.Minute
	} else {
		return 2 * time.Minute
	}
}
func GetEvictLeaderCount(tc resource.TiDBCluster) int {
	evictScheduler := Try(oltppkg.ExecPdCtl(tc, "scheduler config evict-leader-scheduler")).(string)
	r, _ := regexp.Compile("start-key")
	matchIndex := r.FindAllStringIndex(evictScheduler, -1)
	log.Info("Get evict leader scheduler number ", zap.Int("EvictSchedulerNum", len(matchIndex)))
	return len(matchIndex)
}
func NoPausedSchedulerCheck(tc resource.TiDBCluster) {
	out := Try(oltppkg.ExecPdCtl(tc, "scheduler show --status paused")).(string)
	Expect(strings.Contains(out, "balance-leader-scheduler")).Should(BeFalse())
	Expect(strings.Contains(out, "balance-hot-region-scheduler")).Should(BeFalse())
	Expect(strings.Contains(out, "balance-region-scheduler")).Should(BeFalse())
}
func ResumeAllSchdulers(tc resource.TiDBCluster) {
	oltppkg.ExecPdCtl(tc, "scheduler resume balance-leader-scheduler")
	oltppkg.ExecPdCtl(tc, "scheduler resume balance-hot-region-scheduler")
	oltppkg.ExecPdCtl(tc, "scheduler resume balance-region-scheduler")
}
func GetPdConfig(tc resource.TiDBCluster, key string) (string, error) {
	bytes := []byte(Try(oltppkg.ExecPdCtl(tc, "config show all")).(string))
	value := gjson.GetBytes(bytes, key).Value()
	switch reflect.TypeOf(value).Name() {
	case "string":
		return value.(string), nil
	case "bool":
		return strconv.FormatBool(value.(bool)), nil
	case "int64":
		return strconv.FormatInt(value.(int64), 10), nil
	case "float64":
		return strconv.FormatFloat(value.(float64), 'f', -1, 64), nil
	default:
		return "", fmt.Errorf("can't accept error type")
	}
}
func CheckIfPdConfigResetToDefault(tc resource.TiDBCluster) error {
	defaultConfig := map[string]int{
		"max-snapshot-count":     64,
		"max-pending-peer-count": 64,
		"max-merge-region-size":  20,
		"max-merge-region-keys":  200000,
		"leader-schedule-limit":  4,
		"region-schedule-limit":  2048,
	}
	for k, v := range defaultConfig {
		acutal, err := GetPdConfig(tc, fmt.Sprintf("schedule.%s", k))
		log.Info(fmt.Sprintf("compare %s, actual: %s, expect:%d", k, acutal, v))
		Expect(err).Should(BeNil())
		Expect(strconv.Atoi(acutal)).Should(Equal(v))
	}
	return nil
}