package pdrel

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

	. "github.com/onsi/gomega"
	"github.com/pingcap/endless/pkg/chaos"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/pkg/workloadnode"
	"github.com/pingcap/endless/pkg/workloadnode/cluster"
	"github.com/pingcap/endless/pkg/workloadnode/tiup"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	k8swait "k8s.io/apimachinery/pkg/util/wait"
)

func GetPdIngoreErrors(caseCmdParms oltppkg.CaseCmdParms, runType string) []string {
	var ignoreErrors []string
	if strings.HasPrefix(caseCmdParms.WorkloadType.Type[0], "tpcc") {
		if caseCmdParms.TpccParms.WorkloadCheckIgnoreErrors == "" {
			ignoreErrors = strings.Split(caseCmdParms.TpccParms.IgnoreErrors, ",")
		} else {
			ignoreErrors = strings.Split(caseCmdParms.TpccParms.WorkloadCheckIgnoreErrors, ",")
		}
	} else if strings.HasPrefix(caseCmdParms.WorkloadType.Type[0], "ycsb") {
		if caseCmdParms.YcsbParms.WorkloadCheckIgnoreErrors == "" {
			ignoreErrors = strings.Split(caseCmdParms.YcsbParms.IgnoreErrors, ",")
		} else {
			ignoreErrors = strings.Split(caseCmdParms.YcsbParms.WorkloadCheckIgnoreErrors, ",")
		}
	} else { //sysbench or others
		if caseCmdParms.SysbenchParms.WorkloadCheckIgnoreErrors == "" {
			ignoreErrors = strings.Split(caseCmdParms.SysbenchParms.IgnoreErrors, ",")
		} else {
			ignoreErrors = strings.Split(caseCmdParms.SysbenchParms.WorkloadCheckIgnoreErrors, ",")
		}
	}
	//Error 9002: TiKV server timeout <- remove, for will hit when workload is high
	//9001: PD server timeout
	//8027: Information schema is out of date: schema failed to update in 1 lease
	pdRelIgnoreErrors := append(ignoreErrors, "9001")
	switch runType {
	case "fun":
		log.Info("fun ignoreErrors", zap.Strings("ignoreErrors", ignoreErrors))
		return ignoreErrors
	case "rel":
		log.Info("rel ignoreErrors", zap.Strings("ignoreErrors", ignoreErrors))
		return pdRelIgnoreErrors
	default:
		log.Info("now ignoreErrors", zap.Strings("ignoreErrors", ignoreErrors))
		return ignoreErrors
	}

}

func RestartAllNodes(tc resource.TiDBCluster) {
	nodes1 := Try(tc.GetNodes(resource.PD)).([]resource.TiDBComponentNode)
	nodes2 := Try(tc.GetNodes(resource.TiKV)).([]resource.TiDBComponentNode)
	nodes3 := Try(tc.GetNodes(resource.TiDB)).([]resource.TiDBComponentNode)
	nodes := append(append(nodes1, nodes2...), nodes3...)
	var wg sync.WaitGroup
	for _, node := range nodes {
		wg.Add(1)
		go func(node resource.TiDBComponentNode) {
			defer wg.Done()
			startTime := time.Now()
			log.Info(fmt.Sprintf("Now restart node %v", node))
			Try(node.Restart())
			log.Info(fmt.Sprintf("Node:%v restart takes %v", node, time.Since(startTime)))
		}(node)
	}
	wg.Wait()
}

func TikvProfilingUpload(ctx context.Context, tc resource.TiDBCluster, workloadNode resource.WorkloadNode) error {
	profileCmdTpl := `curl "http://%s/debug/pprof/profile?interval=30" > %s`
	cpProfileCmdTpl := "mc alias set qa-minio http://minio.pingcap.net:9000 minioadmin minioadmin --api S3v4;mc cp %s qa-minio/benchbot-profile/%s/%s"

	tikvStatusAddrs, err := tc.PeerURLs(resource.KVStatus)
	if err != nil {
		return err
	}
	for _, tikvAddr := range tikvStatusAddrs {
		timestamp := strconv.FormatInt(time.Now().UnixNano()/1e6, 10)
		filename := fmt.Sprintf("%s_profiling_%s.zip", tikvAddr.Hostname(), timestamp)
		cmdStr := fmt.Sprintf(profileCmdTpl, tikvAddr.Host, fmt.Sprintf("/tmp/%s", filename))
		cmd := workloadnode.CommandContext(ctx, cmdStr)
		Try(cmd.Run(workloadNode))
		cmdStr = fmt.Sprintf(cpProfileCmdTpl, "/tmp/"+filename, strings.Split(tikvAddr.Hostname(), ".")[2], filename)
		cpProfileCmd := workloadnode.CommandContext(ctx, cmdStr)
		Try(cpProfileCmd.Run(workloadNode))
	}
	return nil
}

func GetCaseRunTime() oltppkg.RunTimePara {
	runTimeParms := oltppkg.RunTimePara{
		TotalRunTime: 30 * time.Minute,
		WarmUpTime:   10 * time.Minute,
		RecoveryTime: 10 * time.Minute,
		RunInterval:  30 * time.Minute,
	}
	return runTimeParms
}

func GetPdFaultConditionFun(ctx context.Context, tc resource.TiDBCluster, name string, isLeader bool,
	experimentSpec resource.ExperimentSpec,
	faultDuration time.Duration, RTO time.Duration,
) k8swait.ConditionWithContextFunc {
	if isLeader {
		switch experimentSpec.Kind() {
		case "failure", "network-loss":
			RTO = 2 * time.Minute
		case "network-partition", "network-delay", "io-delay", "time-chaos":
			RTO = faultDuration
		}
	} else {
		RTO = 0 * time.Minute
	}

	conditionWrapper := func(ctx context.Context) (bool, error) {
		c := chaos.Chaos{
			Name:                  name,
			Tc:                    tc,
			ExperimentSpec:        experimentSpec,
			SelectorsRetainPolicy: chaos.RetainRandomly(1),
			SelectorOpts: resource.TiDBClusterSelectorOpts{
				TiDBClusterPDSelectorSpec: &resource.TiDBClusterPDSelectorSpec{
					IsLeader: oltppkg.ToBoolPtr(isLeader),
				},
			},
		}
		faultDuration := oltppkg.GetRandDuration(1*time.Minute, 10)
		log.Info("Courrent pd leader", zap.String("pd leader name", oltppkg.GetPdLeaderName(tc)))
		err := oltppkg.InjectFaultByChaos(&c, ctx, faultDuration)
		if err != nil {
			return true, err
		}
		oltppkg.WaitFor(ctx, "waiting for workload recover", RTO)
		log.Info("Courrent pd leader", zap.String("pd leader name", oltppkg.GetPdLeaderName(tc)))
		oltppkg.RepeatQpsDurCheckWithAbsTime(ctx, 5*time.Minute, time.Now().Add(-RTO).Add(-5*time.Minute), tc, 0.8)
		return false, nil
	}
	return conditionWrapper
}

func (p *PdRelOptions) GetPdFaultConditionFun(ctx context.Context, tc resource.TiDBCluster,
) k8swait.ConditionWithContextFunc {
	experimentSpec := p.GetPdRelExpriment()
	conditionWrapper := func(ctx context.Context) (bool, error) {
		c := chaos.Chaos{
			Name:                  p.FaultType,
			Tc:                    tc,
			ExperimentSpec:        experimentSpec,
			SelectorsRetainPolicy: chaos.RetainRandomly(1),
			SelectorOpts: resource.TiDBClusterSelectorOpts{
				TiDBClusterPDSelectorSpec: &resource.TiDBClusterPDSelectorSpec{
					IsLeader: oltppkg.ToBoolPtr(p.IsLeader),
				},
			},
		}
		startTime := time.Now().Add(p.WarmUpTime / 2)
		oltppkg.WaitFor(ctx, "waiting for workload running", p.WarmUpTime)
		oltppkg.LogAndBy("Inject pd chaos")

		log.Info("Courrent pd leader", zap.String("pd leader name", oltppkg.GetPdLeaderName(tc)))
		var wg sync.WaitGroup
		wg.Add(1)
		go func() {
			defer wg.Done()
			Try(oltppkg.InjectFaultByChaos(&c, ctx, p.FaultDuration))
		}()
		oltppkg.WaitFor(ctx, "waiting for workload recover", p.RTO)
		//log.Info("Courrent pd leader", zap.String("pd leader name", oltppkg.GetPdLeaderName(tc)))
		oltppkg.RepeatQpsDurCheckWithAbsTime(ctx, 5*time.Minute, time.Now().Add(-p.RTO).Add(-p.WarmUpTime/2), tc, 0.5)

		oltppkg.WaitFor(ctx, "waiting for recover time", p.RecoveryTime)
		//log.Info("Courrent pd leader", zap.String("pd leader name", oltppkg.GetPdLeaderName(tc)))
		oltppkg.RepeatQpsDurCheckWithAbsTime(ctx, 5*time.Minute, startTime, tc, 0.2)

		wg.Wait()
		return false, nil
	}
	return conditionWrapper
}

func (p *PdRelOptions) GetPdRelExpriment() resource.ExperimentSpec {
	var experimentSpec resource.ExperimentSpec
	switch p.FaultType {
	case "failure":
		experimentSpec = &resource.FailureExperimentSpec{}
	case "io-delay":
		experimentSpec = &resource.IODelaySpec{
			VolumePath: "/var/lib/pd",
			Delay:      oltppkg.GetRandIntStr(100, 500) + "ms",
			Percent:    int64(oltppkg.GetRandInt(10, 90)),
		}
	case "network-loss-100":
		experimentSpec = &resource.NetworkLossSpec{
			Loss:        "100",
			Correlation: "70"}
	case "network-loss":
		experimentSpec = &resource.NetworkLossSpec{
			Duration:    oltppkg.GetRandIntStr(1, 3) + "s",
			Scheduler:   &resource.ExperimentScheduler{Cron: "@every " + oltppkg.GetRandIntStr(5, 60) + "s"},
			Loss:        "100",
			Correlation: "70"}
	case "network-latency":
		experimentSpec = &resource.NetworkLatencySpec{
			Latency:     "5ms", //oltppkg.GetRandIntStr(9, 10) + "ms",
			Correlation: "25",
			Jitter:      "0ms", //"2ms"
		}
	case "time-offset":
		timeOffset := fmt.Sprintf("%dms", oltppkg.GetRandInt(-1000, 1000))
		experimentSpec = &resource.TimeChaosSpec{TimeOffset: timeOffset}
	default:
		log.Warn("can't support type")
		return nil
	}
	return experimentSpec
}
func (p *PdRelOptions) GetPdFaultConditionFunTiup(ctx context.Context, ctxCore core.TestContext, tc *tiup.TiUPCluster,
) k8swait.ConditionWithContextFunc {
	experiment := p.GetPdRelExpriment()
	conditionWrapper := func(ctx context.Context) (bool, error) {
		startTime := time.Now().Add(p.WarmUpTime / 2)
		oltppkg.WaitFor(ctx, "waiting for workload running", p.WarmUpTime)
		oltppkg.LogAndBy("Inject pd chaos")

		log.Info("Courrent pd leader", zap.String("pd leader name", oltppkg.GetPdLeaderName(tc)))
		var wg sync.WaitGroup
		wg.Add(1)
		go func() {
			defer wg.Done()
			_, pdLeaderNode := GetTiupPdLeaderNode(ctxCore, tc)
			oltppkg.InjectChaosToWorkloadNode(ctx, pdLeaderNode, experiment, p.FaultDuration)
		}()
		oltppkg.WaitFor(ctx, "waiting for workload recover", p.RTO)
		log.Info("Courrent pd leader", zap.String("pd leader name", oltppkg.GetPdLeaderName(tc)))
		oltppkg.RepeatQpsDurCheckWithAbsTime(ctx, 5*time.Minute, time.Now().Add(-p.RTO).Add(-p.WarmUpTime/2), tc, 0.5)

		oltppkg.WaitFor(ctx, "waiting for recover time", p.RecoveryTime)
		oltppkg.RepeatQpsDurCheckWithAbsTime(ctx, 5*time.Minute, startTime, tc, 0.2)

		wg.Wait()
		return false, nil
	}
	return conditionWrapper
}
func GetTiupPdLeaderNode(ctx core.TestContext, tc *tiup.TiUPCluster) (string, resource.WorkloadNode) {
	pdLeader := oltppkg.GetPdLeaderName(tc)
	matchStr := regexp.MustCompile(`pd-(.*)-peer-.*`).FindStringSubmatch(pdLeader)
	Expect(len(matchStr)).Should(Equal(2))
	pdLeaderName := matchStr[1]
	log.Info("pd leader", zap.String("pd leader peer", pdLeaderName))
	pdLeaderNode := ctx.Resource(pdLeaderName).(resource.WorkloadNode)
	return pdLeaderName, pdLeaderNode
}
func TiupDownPdLeader(ctx core.TestContext, tc *tiup.TiUPCluster, faultDuration time.Duration) {
	_, pdLeaderNode := GetTiupPdLeaderNode(ctx, tc)
	log.Info("Inject chaos to ", zap.String("workloadName", pdLeaderNode.Handle().Name))
	experimentSpec := &resource.NetworkLossSpec{
		Loss:        "100",
		Correlation: "25",
	}
	chaoId := Try(oltppkg.InjectChaosToWorkloadNode(ctx, pdLeaderNode, experimentSpec, faultDuration)).(resource.ChaosIdentifier)
	defer func() {
		log.Info("Try to delete the chaos")
		Try(pdLeaderNode.ChaosClient().Delete(chaoId))
	}()
	oltppkg.WaitFor(ctx, "waiting for to clean chaos", faultDuration)
	select {
	case <-ctx.Done():
		log.Info("context done ,delete chaos")
	case <-time.After(faultDuration):
		log.Info("time out,delete chaos")
	}
	Try(pdLeaderNode.ChaosClient().Delete(chaoId))
}

func CreateChaosOnWorkloadNode(ctx context.Context, workloadNode resource.WorkloadNode) (resource.ChaosIdentifier, error) {
	selectors, _ := workloadNode.ChaosSelect(resource.WorkloadNodeSelectorOpts{All: true})
	experiment := &resource.NetworkLossSpec{Loss: "100", Correlation: "100"}
	id, err := workloadNode.ChaosClient().Create(
		resource.InjectChaosOptions{
			Selector:       selectors,
			ExperimentSpec: experiment,
		},
	)
	return id, err
}

type PdRelTestParams struct {
	IgnoreErr      []string
	ExperimentSpec resource.ExperimentSpec
}

func PdRelTest(ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
	name string,
	isLeader bool,
	experimentSpec resource.ExperimentSpec,
	RTO time.Duration,
	commonCheckParams ...interface{}) {
	runTimeParms := GetCaseRunTime()
	ctxAll, cancelAll := context.WithTimeout(ctx, runTimeParms.TotalRunTime)
	defer cancelAll()
	g, ctxg := errgroup.WithContext(ctxAll)

	startTime := time.Now().Add(runTimeParms.WarmUpTime / 2)
	log.Info(fmt.Sprintf("%v", commonCheckParams))
	oltppkg.FunCommonCheck(ctxg, tc, g, runTimeParms.TotalRunTime, workloadRunParms.Cmdpr, commonCheckParams...)

	//Inject chaos
	oltppkg.WaitFor(ctxAll, "waiting for workload running", runTimeParms.WarmUpTime)
	oltppkg.LogAndBy("Inject pd chaos")

	conditionWrapper := GetPdFaultConditionFun(ctxg, tc, name, isLeader, experimentSpec, oltppkg.GetRandDuration(1*time.Minute, 5), RTO)

	g.Go(WithGinkgoRecover(func() {
		k8swait.PollImmediateUntilWithContext(ctxg, runTimeParms.RunInterval, conditionWrapper)
	}))

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

	oltppkg.SleepWithCtx(ctx, runTimeParms.RecoveryTime)
	oltppkg.RepeatQpsDurCheckWithAbsTime(ctxg, 5*time.Minute, startTime, tc, 0.2)
}

type PdRelOptions struct {
	FaultType       string
	IsLeader        bool
	RTO             time.Duration
	WarmUpTime      time.Duration
	FaultDuration   time.Duration
	RecoveryTime    time.Duration
	RunTime         time.Duration
	PollingInterval time.Duration
}

func (p *PdRelOptions) PdRelTest(
	ctx core.TestContext,
	tc cluster.BaseCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms) {
	ctxAll, cancelAll := context.WithTimeout(ctx, p.RunTime)
	defer cancelAll()
	g, ctxg := errgroup.WithContext(ctxAll)

	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: false,
		TiKVRestartCheck:         true,
		PDRestartCheck:           false,
		TiDBRestartCheck:         true,
	}
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck:      false,
		Offset:       "2m",
		ToleratePerc: 0.8,
		TolerateDur:  600 * time.Second,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck:      false,
		Offset:       "2m",
		ToleratePerc: 2,
		TolerateDur:  600 * time.Second,
	}
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: GetPdIngoreErrors(caseCmdParms, "rel"),
	}
	// ignore this check to avoid hit issue https://github.com/pingcap/tidb/issues/35963
	tidbPanicAndCriticalErrCheckParms := oltppkg.TidbPanicAndCriticalErrCheckParameter{
		IsPanicCheck:    false,
		IsCriticalCheck: false,
	}
	oltppkg.FunCommonCheck(ctxg, tc, g, p.RunTime, workloadRunParms.Cmdpr, restartCheckParms, qpsJitterCheckParms, durJitterCheckParms, workloadCheckParms, tidbPanicAndCriticalErrCheckParms)

	//Inject chaos
	var conditionWrapper k8swait.ConditionWithContextFunc
	if _, ok := tc.(resource.TiDBCluster); ok {
		conditionWrapper = p.GetPdFaultConditionFun(ctxg, tc.(resource.TiDBCluster))
	} else {
		conditionWrapper = p.GetPdFaultConditionFunTiup(ctxg, ctx, tc.(*tiup.TiUPCluster))
	}

	g.Go(WithGinkgoRecover(func() {
		k8swait.PollImmediateUntilWithContext(ctxg, p.PollingInterval, conditionWrapper)
	}))

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