package pdfun

import (
	"context"
	"fmt"
	"strings"
	"time"

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

func OLTP_SHUFFLE_SCHEDULER_STABILITY_001(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	oltppkg.LogAndBy("Add shuttle schedule more than 10 hours.")
	client := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	Try(client.SetConfig("split-merge-interval", "10m"))
	defer Try(client.SetConfig("split-merge-interval", "30m"))
	Try(oltppkg.ExecPdCtl(tc, "config set merge-schedule-limit 80"))
	defer Try(oltppkg.ExecPdCtl(tc, "config set merge-schedule-limit 8"))

	runTimePara := oltppkg.NewRunTimePara(10*time.Hour, 20*time.Minute, 0*time.Second, 20*time.Minute)
	ctxAll, cancelAll := context.WithTimeout(ctx, runTimePara.TotalRunTime)
	defer cancelAll()
	g, ctxg := errgroup.WithContext(ctxAll)

	oltppkg.FunCommonCheck(ctxg, tc, g, runTimePara.TotalRunTime, workloadRunParms.Cmdpr)

	startTime := time.Now()
	oltppkg.WaitFor(ctxg, "Workload warmup", runTimePara.WarmUpTime)

	g.Go(WithGinkgoRecover(func() {
		oltppkg.MonitorWorkloadBalance(ctxg, tc, caseCmdParms.WorkloadType.Type[0], g, runTimePara.TotalRunTime-runTimePara.WarmUpTime, 0.8)
	}))
	g.Go(WithGinkgoRecover(func() {
		oltppkg.WaitFor(ctxg, "waiting workload run 5 hours to start RandomScheduler", 5*time.Hour)
		oltppkg.SetPDRandomScheduler(tc, "add")
	}))
	defer oltppkg.SetPDRandomScheduler(tc, "remove")

	workloadType := caseCmdParms.WorkloadType.Type[0]
	var dbName string
	if strings.HasPrefix(workloadType, "sysbench_") {
		dbName = caseCmdParms.SysbenchParms.Db
	} else if strings.HasPrefix(workloadType, "tpcc") {
		dbName = caseCmdParms.TpccParms.Db
	} else {
		panic("invalid workloadType")
	}

	g.Go(WithGinkgoRecover(func() {
		oltppkg.SplitTablesRepeatly(ctxg, tc, dbName, 5*time.Minute)
	}))

	Expect(g.Wait()).Should(BeNil())
	oltppkg.WaitFor(ctxg, "Workload warmup", runTimePara.RecoveryTime)
	oltppkg.LogAndBy("Check qps with the time before begin to split the region.")
	oltppkg.RepeatQpsDurCheckWithAbsTime(ctx, runTimePara.WarmUpTime, startTime, tc, 0.9)
}

func OLTP_DISK_FULL_STABILITY_001(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadNode resource.WorkloadNode,
	skipAfterEachCheck chan bool,
) {
	workloadRunParms := oltppkg.WorkloadRunParms{}
	if caseCmdParms.SkipRunWorkload {
		workloadRunParms.Init(ctx, tc, workloadNode)
	}
	var runTime time.Duration
	if caseCmdParms.RunTime.TotalRunTime != 0 {
		runTime = caseCmdParms.RunTime.TotalRunTime
	} else {
		runTime = 24 * time.Hour
	}
	var recoverTime time.Duration
	if caseCmdParms.RunTime.RecoveryTime != 0 {
		recoverTime = caseCmdParms.RunTime.TotalRunTime
	} else {
		recoverTime = 30 * time.Minute
	}
	log.Info("Will run ", zap.Duration("total run time", runTime))
	skipAfterEachCheck <- true

	ctxAll, cancelAll := context.WithTimeout(ctx, runTime)
	defer cancelAll()
	stopRunningFlag := make(chan bool)

	Try(oltppkg.ExecPdCtl(tc, "store limit all 200"))
	defer func() {
		Try(oltppkg.ExecPdCtl(tc, "store limit all 15"))
	}()

	//Start oom and output check before all.
	oltppkg.LogAndBy("Step 1:Start oom and output check before all.")
	g, ctxg := errgroup.WithContext(ctxAll)
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: strings.Split(caseCmdParms.SysbenchParms.IgnoreErrors, ","),
	}
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck: false,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck: false,
	}
	applyDbMutexLockDurCheckParms := oltppkg.ApplyDbMutexLockDurCheckParameters{
		TolerateMaxDur: 2.0, //2.0(s)
	}
	oltppkg.FunCommonCheck(ctxg, tc, g, runTime, workloadRunParms.Cmdpr, workloadCheckParms, qpsJitterCheckParms, durJitterCheckParms, applyDbMutexLockDurCheckParms)

	storeSize, sumSize, err := oltppkg.GetStoreSizeByType(tc, "used_size")
	Expect(err).Should(Succeed())
	log.Info(fmt.Sprintf("Store Size and sumSize%v %v before run insert.", storeSize, sumSize))

	g2, ctxg2 := errgroup.WithContext(ctxg)
	g2.Go(WithGinkgoRecover(func() {
		k8swait.PollImmediateUntilWithContext(ctxg2, 5*time.Minute, getRunInsertFunc(ctxg2, tc, workloadRunParms, recoverTime))
	}))
	g2.Go(WithGinkgoRecover(func() {
		k8swait.PollImmediateUntilWithContext(ctxg2, 1*time.Minute, getStopWorkloadFunc(ctxg2, tc, workloadRunParms, stopRunningFlag))
	}))
	g2.Go(WithGinkgoRecover(func() {
		k8swait.PollImmediateUntilWithContext(ctxg2, 1*time.Minute, getDiskFullCheckFunc(ctxg2, tc, stopRunningFlag))
	}))
	Expect(g2.Wait()).Should(BeNil())
	log.Info("Test finished, cancel all to stop checks.")
	workloadRunParms.Cmdpw.Close()
	workloadRunParms.Cmdpr.Close()
	cancelAll()
	Expect(g.Wait()).Should(BeNil())
	log.Info("Done")
}

func OLTP_SPLIT_MERGE_STABILITY_REL_001(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	oltppkg.LogAndBy("Add shuttle schedule more than 10 hours.")
	client := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	Try(client.SetConfig("split-merge-interval", "10m"))
	Try(oltppkg.ExecPdCtl(tc, "config set merge-schedule-limit 80"))
	defer func() {
		Try(client.SetConfig("split-merge-interval", "30m"))
		Try(oltppkg.ExecPdCtl(tc, "config set merge-schedule-limit 8"))

	}()

	runTimePara := oltppkg.NewRunTimePara(10*time.Hour, 10*time.Minute, oltppkg.GetRandDuration(time.Minute, 10), 10*time.Minute)
	ctxAll, cancelAll := context.WithTimeout(ctx, runTimePara.TotalRunTime)
	defer cancelAll()
	g, ctxg := errgroup.WithContext(ctxAll)

	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck:      false,
		Offset:       "2m",
		ToleratePerc: 0.9,
		TolerateDur:  120 * time.Second,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck: false,
	}
	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: false,
		TiKVRestartCheck:         true,
		PDRestartCheck:           false,
		TiDBRestartCheck:         true,
	}

	oltppkg.FunCommonCheck(ctxg, tc, g, runTimePara.TotalRunTime, workloadRunParms.Cmdpr, qpsJitterCheckParms, durJitterCheckParms, restartCheckParms)

	startTime := time.Now().Add(runTimePara.WarmUpTime / 2)
	oltppkg.WaitFor(ctxg, "Workload warmup", runTimePara.WarmUpTime)

	g.Go(WithGinkgoRecover(func() {
		workloadType := caseCmdParms.WorkloadType.Type[0]
		var dbName string
		if strings.HasPrefix(workloadType, "sysbench_") {
			dbName = caseCmdParms.SysbenchParms.Db
		} else if strings.HasPrefix(workloadType, "tpcc") {
			dbName = caseCmdParms.TpccParms.Db
		} else {
			panic("invalid workloadType")
		}
		oltppkg.SplitTablesRepeatly(ctxg, tc, dbName, 5*time.Minute)
	}))

	g.Go(WithGinkgoRecover(func() {
		oltppkg.MonitorWorkloadBalance(ctxg, tc, caseCmdParms.WorkloadType.Type[0], g, runTimePara.TotalRunTime-runTimePara.WarmUpTime, 2.0)
	}))

	g.Go(WithGinkgoRecover(func() {
		c := Try(oltppkg.InitChaos(tc, resource.PD, "network loss", &resource.NetworkLossSpec{Loss: "20", Correlation: "50"}, oltppkg.ToBoolPtr(true))).(chaos.Chaos)
		c.RunPeriodically(ctxg, "@every 10m", runTimePara.FaultDuration)
		oltppkg.WaitFor(ctxg, "wait to clean network loss chaos.", runTimePara.TotalRunTime/2)
		c.Clean()
	}))
	g.Go(WithGinkgoRecover(func() {
		oltppkg.WaitFor(ctxg, "wait network loss inject finished.", runTimePara.TotalRunTime/2)
		c := Try(oltppkg.InitChaos(tc, resource.PD, "failure", &resource.FailureExperimentSpec{}, oltppkg.ToBoolPtr(true))).(chaos.Chaos)
		c.RunPeriodically(ctxg, "@every 15m", runTimePara.FaultDuration)
	}))

	g.Go(WithGinkgoRecover(func() {
		oltppkg.WaitFor(ctxg, "waiting workload run 5 hours to start RandomScheduler", runTimePara.TotalRunTime/2)
		oltppkg.SetPDRandomScheduler(tc, "add")
	}))
	defer oltppkg.SetPDRandomScheduler(tc, "remove")

	Expect(g.Wait()).Should(BeNil())
	oltppkg.WaitFor(ctx, "Workload recovery", runTimePara.RecoveryTime)
	oltppkg.LogAndBy("Check qps with the time before begin to split the region.")
	oltppkg.RepeatQpsDurCheckWithAbsTime(ctx, runTimePara.WarmUpTime, startTime, tc, 0.9)

	if oltppkg.CompareVersionWithCurrent(ctx, tc, "5.4.0") >= 0 {
		log.Info("Check hot region size if current version is higher than 5.4.0")
		//CheckHotRegionSize(ctx, tc)
	}
}