package tikvrel

import (
	"context"
	"errors"
	"fmt"
	"io"
	"strings"
	"time"

	_ "github.com/go-sql-driver/mysql"
	. "github.com/onsi/gomega"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"

	"github.com/pingcap/endless/pkg/chaos"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/matcher/metrics"
	. "github.com/pingcap/endless/pkg/util"
	"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"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"
)

type TiKVRelOptions struct {
	FaultType     string
	FaultNum      int
	RTO           time.Duration
	WarmUpTime    time.Duration
	FaultDuration time.Duration
	RecoveryTime  time.Duration
}

type faultInjectNodeOrChaos struct {
	falutNode       resource.TiDBComponentNode
	workloadNode    resource.WorkloadNode
	c               chaos.Chaos
	falutInstanceId int
}

func (tikvRelOpts *TiKVRelOptions) GetRunTimeAndCheckOffset(ctx context.Context, tc resource.MySQLCompatible, caseCmdParms oltppkg.CaseCmdParms) (time.Duration, string) {
	if caseCmdParms.RunTime.WarmUpTime != 0 && caseCmdParms.RunTime.FaultDuration != 0 && caseCmdParms.RunTime.RecoveryTime != 0 {
		tikvRelOpts.WarmUpTime = caseCmdParms.RunTime.WarmUpTime
		tikvRelOpts.FaultDuration = caseCmdParms.RunTime.FaultDuration
		tikvRelOpts.RecoveryTime = caseCmdParms.RunTime.RecoveryTime
	}

	if tikvRelOpts.FaultType == "restart" {
		tikvRelOpts.FaultDuration = 5 * time.Minute
	}
	runTime := tikvRelOpts.WarmUpTime + tikvRelOpts.FaultDuration + tikvRelOpts.RecoveryTime

	// tikv may be not trigger an election if "network-delay" or "IODelay" occur
	// know issue INCALL-113 (https://github.com/tikv/client-go/issues/521), avoid this check due to issue do not fixed
	tidbReplicaRead := Try(oltppkg.GetTidbVariable(ctx, tc, "tidb_replica_read")).(string)
	var recoveryCheckOffset time.Duration
	if strings.Contains(tikvRelOpts.FaultType, "network-delay") || strings.Contains(tikvRelOpts.FaultType, "IODelay") || tidbReplicaRead == "follower" {
		recoveryCheckOffset = tikvRelOpts.WarmUpTime + tikvRelOpts.FaultDuration
		if tikvRelOpts.RTO == 0 {
			tikvRelOpts.RTO = tikvRelOpts.FaultDuration + 2*time.Minute
		}
	} else {
		recoveryCheckOffset = tikvRelOpts.WarmUpTime
		if tikvRelOpts.RTO == 0 {
			tikvRelOpts.RTO = 2 * time.Minute
		}
	}
	recoveryCheckOffsetStr := strings.Split(recoveryCheckOffset.String(), "m")[0] + "m"

	log.Info("TiKVRelOpts",
		zap.Duration("WarmUpTime", tikvRelOpts.WarmUpTime),
		zap.Duration("FaultDuration", tikvRelOpts.FaultDuration),
		zap.Duration("RecoveryTime", tikvRelOpts.RecoveryTime),
		zap.String("FaultType", tikvRelOpts.FaultType),
		zap.Int("FaultNum", tikvRelOpts.FaultNum),
		zap.Duration("RTO", tikvRelOpts.RTO))

	return runTime, recoveryCheckOffsetStr
}

func (tikvRelOpts *TiKVRelOptions) TiKVFaultTest(
	ctx core.TestContext,
	tc resource.MySQLCompatible,
	pr *io.PipeReader,
	caseCmdParms oltppkg.CaseCmdParms,
) {
	oltppkg.LogAndBy("minority replicas failed testing")
	faultInjectNodeOrChaos1, faultInjectNodeOrChaos2, instanceIdList, falutInstanceIdList := tikvRelOpts.GetFaultInjectNodeOrChaos(ctx, tc)
	normalInstanceIdList := tikvRelOpts.getNormalTiKVInstance(instanceIdList, falutInstanceIdList)
	waitForRaftLogCatchUp := 5 * time.Hour
	waitForBalance := 3 * time.Hour
	minRunTime, recoveryCheckOffsetStr := tikvRelOpts.GetRunTimeAndCheckOffset(ctx, tc, caseCmdParms)
	maxRunTime := minRunTime + waitForRaftLogCatchUp + waitForBalance
	ctxAll, cancel := context.WithTimeout(ctx, maxRunTime)
	defer cancel()
	g, gctx := errgroup.WithContext(ctxAll)

	tikvRelOpts.CommonCheckDuringCaseRunning(gctx, tc, g, maxRunTime, pr, caseCmdParms)
	if tikvRelOpts.FaultType == "failure" || tikvRelOpts.FaultType == "restart" {
		log.Info("normal TiKV restart check during the whole process of running the case ")
		tikvRelOpts.tikvInstanceRestartCheck(gctx, tc, g, maxRunTime, normalInstanceIdList)
		log.Info("the tikv will be inject failure or restart check during warm up")
		tikvRelOpts.tikvInstanceRestartCheck(gctx, tc, g, tikvRelOpts.WarmUpTime, falutInstanceIdList)
	}

	oltppkg.WaitFor(gctx, "Warm Up before inject fault", tikvRelOpts.WarmUpTime)

	log.Info("Fault Info", zap.Ints("all instanceId", instanceIdList),
		zap.Ints("will be injected fault InstanceId", falutInstanceIdList),
		zap.String("Fault Type", tikvRelOpts.FaultType))

	tikvRelOpts.InjectFaultAccordingToNum(gctx, tc, g, faultInjectNodeOrChaos1, faultInjectNodeOrChaos2)

	g.Go(WithGinkgoRecover(func() {
		oltppkg.WaitFor(gctx, "Wait QPS recover after inject fault.", tikvRelOpts.RTO)
		tidbReplicaRead := Try(oltppkg.GetTidbVariable(ctx, tc, "tidb_replica_read")).(string)
		if !strings.Contains(tikvRelOpts.FaultType, "network-delay") && !strings.Contains(tikvRelOpts.FaultType, "IODelay") && tidbReplicaRead != "follower" {
			log.Info("start qps check before fault recover,  expect qps to recover to 50% of pre-failure")
			ConsistentlyWithMetrics(tc, tikvRelOpts.FaultDuration-tikvRelOpts.RTO).
				ShouldNot(QPSDropMoreThanOffset(0.5, recoveryCheckOffsetStr).For(30 * time.Second).WithCtx(gctx))
			log.Info("start qps check after fault recover,  do not expect qps to drop more than 50% for 300s")
			// know issue https://github.com/tikv/tikv/issues/12259 and incall 106, so change QPS jitter duration from 30s to 300s
			ConsistentlyWithMetrics(tc, maxRunTime).
				ShouldNot(QPSDropMoreThanOffset(0.5, recoveryCheckOffsetStr).For(300 * time.Second).WithCtx(gctx))
		} else {
			if tidbReplicaRead == "follower" {
				// know issue INCALL-113 (https://github.com/tikv/client-go/issues/521), avoid this check due to issue do not fixed
				log.Info("start qps check after fault recover, do not expect qps to drop more than 50% for 300s")
				ConsistentlyWithMetrics(tc, maxRunTime).
					ShouldNot(QPSDropMoreThanOffset(0.5, recoveryCheckOffsetStr).For(300 * time.Second).WithCtx(gctx))
			} else {
				log.Info("start qps check after fault recover, do not expect qps to drop more than 30% for 30s")
				ConsistentlyWithMetrics(tc, maxRunTime).
					ShouldNot(QPSDropMoreThanOffset(0.3, recoveryCheckOffsetStr).For(30 * time.Second).WithCtx(gctx))
			}
		}
	}))

	oltppkg.WaitFor(gctx, "wait for fault recover.", tikvRelOpts.FaultDuration+3*time.Minute)

	if tikvRelOpts.FaultType == "failure" || tikvRelOpts.FaultType == "restart" {
		log.Info("TiKV restart check after failure or restart recovery")
		tikvRelOpts.tikvInstanceRestartCheck(gctx, tc, g, maxRunTime, falutInstanceIdList)
	}

	tikvRelOpts.WaitClusterStatusRecover(gctx, tc, waitForRaftLogCatchUp, waitForBalance)
	oltppkg.WaitFor(gctx, "Recovery observation period", tikvRelOpts.RecoveryTime)
	cancel()

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

func (tikvRelOpts *TiKVRelOptions) GetFaultInjectNodeOrChaos(ctx core.TestContext, tc resource.MySQLCompatible) (faultInjectNodeOrChaos, faultInjectNodeOrChaos, []int, []int) {
	var faultInjectNodeOrChaos1 faultInjectNodeOrChaos
	var faultInjectNodeOrChaos2 faultInjectNodeOrChaos
	var instanceIdList []int
	var falutInstanceIdList []int
	c := tikvRelOpts.GetExperimentSpec()
	tcType := fmt.Sprintf("%T", tc)
	fmt.Printf("tcType:%s\n", tcType)
	switch tcType {
	case "*k8s.TiDBCluster":
		instanceType := resource.TiKV
		instanceIdList = oltppkg.GetInstanceIdList(tc.(resource.TiDBCluster), instanceType)
		tikvNum := len(instanceIdList)
		faultInjectNodeOrChaos1.falutInstanceId = oltppkg.GetRandInt(0, tikvNum)
		faultInjectNodeOrChaos2.falutInstanceId = (faultInjectNodeOrChaos1.falutInstanceId + 1) % tikvNum
		faultInjectNodeOrChaos1.falutNode = Try(tc.(resource.TiDBCluster).GetNodes(instanceType)).([]resource.TiDBComponentNode)[faultInjectNodeOrChaos1.falutInstanceId]
		faultInjectNodeOrChaos2.falutNode = Try(tc.(resource.TiDBCluster).GetNodes(instanceType)).([]resource.TiDBComponentNode)[faultInjectNodeOrChaos2.falutInstanceId]
		faultInjectNodeOrChaos1.c = Try(oltppkg.InitChaos(tc.(resource.TiDBCluster), instanceType, tikvRelOpts.FaultType, c.ExperimentSpec, &faultInjectNodeOrChaos1.falutInstanceId)).(chaos.Chaos)
		faultInjectNodeOrChaos2.c = Try(oltppkg.InitChaos(tc.(resource.TiDBCluster), instanceType, tikvRelOpts.FaultType, c.ExperimentSpec, &faultInjectNodeOrChaos2.falutInstanceId)).(chaos.Chaos)
	case "*tiup.TiUPCluster":
		_, tikvNodeList := oltppkg.GetComponentMapByType(ctx, tc.(*tiup.TiUPCluster), "tikv", "up")
		tikvNum := len(tikvNodeList)
		for i := 0; i < tikvNum; i++ {
			instanceIdList = append(instanceIdList, i)
		}
		faultInjectNodeOrChaos1.falutInstanceId = oltppkg.GetRandInt(0, tikvNum)
		faultInjectNodeOrChaos2.falutInstanceId = (faultInjectNodeOrChaos1.falutInstanceId + 1) % tikvNum
		faultInjectNodeOrChaos1.workloadNode = tikvNodeList[faultInjectNodeOrChaos1.falutInstanceId]
		faultInjectNodeOrChaos2.workloadNode = tikvNodeList[faultInjectNodeOrChaos2.falutInstanceId]
		faultInjectNodeOrChaos1.c = c
		faultInjectNodeOrChaos2.c = c

	default:
		Expect(errors.New("invalid tc type")).Should(Succeed())
	}
	falutInstanceIdList = append(falutInstanceIdList, faultInjectNodeOrChaos1.falutInstanceId)
	if tikvRelOpts.FaultNum == 2 {
		falutInstanceIdList = append(falutInstanceIdList, faultInjectNodeOrChaos2.falutInstanceId)
	}
	return faultInjectNodeOrChaos1, faultInjectNodeOrChaos2, instanceIdList, falutInstanceIdList
}

func (tikvRelOpts *TiKVRelOptions) InjectFaultAccordingToNum(
	gctx context.Context,
	tc resource.MySQLCompatible,
	g *errgroup.Group,
	faultInjectNodeOrChaos1 faultInjectNodeOrChaos,
	faultInjectNodeOrChaos2 faultInjectNodeOrChaos,
) {
	g.Go(WithGinkgoRecover(func() {
		if tikvRelOpts.FaultType == "restart" {
			Try(faultInjectNodeOrChaos1.falutNode.Restart())
			oltppkg.WaitFor(gctx, "Wait for tikv recovery.", tikvRelOpts.FaultDuration)
		} else {
			tikvRelOpts.InjectFault(gctx, tc, faultInjectNodeOrChaos1.c, faultInjectNodeOrChaos1.workloadNode, tikvRelOpts.FaultDuration)
		}
	}))

	if tikvRelOpts.FaultNum == 2 {
		g.Go(WithGinkgoRecover(func() {
			if tikvRelOpts.FaultType == "restart" {
				Try(faultInjectNodeOrChaos2.falutNode.Restart())
				oltppkg.WaitFor(gctx, "Wait for tikv recovery.", tikvRelOpts.FaultDuration)
			} else {
				tikvRelOpts.InjectFault(gctx, tc, faultInjectNodeOrChaos2.c, faultInjectNodeOrChaos2.workloadNode, tikvRelOpts.FaultDuration)
			}
		}))
	}
}

func (tikvRelOpts *TiKVRelOptions) InjectFault(ctx context.Context, tc resource.MySQLCompatible, c chaos.Chaos, workloadNode resource.WorkloadNode, faultDuration time.Duration) {
	tcType := fmt.Sprintf("%T", tc)
	fmt.Printf("tcType:%s\n", tcType)
	switch tcType {
	case "*k8s.TiDBCluster":
		err := oltppkg.InjectFaultByChaos(&c, ctx, tikvRelOpts.FaultDuration)
		Expect(err).Should(Succeed())
	case "*tiup.TiUPCluster":
		err := oltppkg.InjectChaosToWorkloadNode(ctx, workloadNode, c.ExperimentSpec, tikvRelOpts.FaultDuration)
		Expect(err).Should(Succeed())
	default:
		Expect(errors.New("invalid tc type")).Should(Succeed())
	}
}

func (tikvRelOpts *TiKVRelOptions) GetExperimentSpec() chaos.Chaos {
	var c chaos.Chaos
	if strings.Contains(tikvRelOpts.FaultType, "failure") {
		c.ExperimentSpec = &resource.FailureExperimentSpec{}
	} else if strings.Contains(tikvRelOpts.FaultType, "network-loss") {
		loss := strings.TrimPrefix(tikvRelOpts.FaultType, "network-loss-loss:")
		if loss == "random" {
			loss = oltppkg.GetRandIntStr(1, 100)
		}
		c.ExperimentSpec = &resource.NetworkLossSpec{
			Loss:        loss,
			Correlation: "25",
		}
	} else if strings.Contains(tikvRelOpts.FaultType, "network-delay") {
		latency := strings.TrimPrefix(tikvRelOpts.FaultType, "network-delay-latency:")
		// https://github.com/tikv/tikv/issues/12855 if latency more then 300ms
		if latency == "random" {
			latency = oltppkg.GetRandIntStr(100, 200) + "ms"
		}
		c.ExperimentSpec = &resource.NetworkLatencySpec{
			Latency:     latency,
			Correlation: "25",
		}
	} else if strings.Contains(tikvRelOpts.FaultType, "IODelay") {
		delay := strings.TrimPrefix(tikvRelOpts.FaultType, "IODelay-delay:")
		if delay == "random" {
			// delay = oltppkg.GetRandIntStr(100, 500) + "ms"
			delay = "10ms"
		}
		c.ExperimentSpec = &resource.IODelaySpec{
			VolumePath: "/var/lib/tikv",
			Path:       "/var/lib/tikv/data/**/*",
			Delay:      delay, //base delay: 40ms
			Percent:    100,
		}
	} else if strings.Contains(tikvRelOpts.FaultType, "restart") {
		log.Info("will test restart, do not need chaos")
	} else {
		Expect(errors.New("invalid faultType")).Should(Succeed())
	}
	return c
}

func (tikvRelOpts *TiKVRelOptions) CommonCheckDuringCaseRunning(
	gctx context.Context,
	tc resource.MySQLCompatible,
	g *errgroup.Group,
	runTime time.Duration,
	pr *io.PipeReader,
	caseCmdParms oltppkg.CaseCmdParms,
) {
	//Error 8027: https://github.com/pingcap/tidb/issues/30258
	workloadCheckParams := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: append(strings.Split(caseCmdParms.TpccParms.IgnoreErrors, ","), "invalid connection", "err dial tcp"),
	}
	qpsJitterCheckParams := oltppkg.QpsJitterCheckParameters{
		IsCheck: false,
	}
	durJitterCheckParams := oltppkg.DurJitterCheckParameters{
		IsCheck: false,
	}
	oomCheckParams := tikvRelOpts.getOomCheckParams()
	restartCheckParams := tikvRelOpts.getRestartCheckParams()
	raftEngineAppendFileCountCheckParams := tikvRelOpts.getRaftEngineGcCheckParams()
	oltppkg.FunCommonCheck(gctx, tc, g, runTime, pr, qpsJitterCheckParams, durJitterCheckParams, restartCheckParams, oomCheckParams, workloadCheckParams, raftEngineAppendFileCountCheckParams)
}

func (tikvRelOpts *TiKVRelOptions) tikvInstanceRestartCheck(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, runTime time.Duration, instanceIdList []int) {
	log.Info("tikvInstanceRestartCheck", zap.Ints("InstanceId", instanceIdList))
	for _, instanceId := range instanceIdList {
		g.Go(WithGinkgoRecover(func() {
			id := instanceId
			instance := fmt.Sprintf("tc-tikv-%d", id)
			log.Info(fmt.Sprintf("%s restart check", instance))
			ConsistentlyWithMetrics(tc, runTime).
				ShouldNot(InstanceRestart(instance).WithCtx(ctx))
			log.Info(fmt.Sprintf("tc-tikv-%d restart check done", id))
		}))
		oltppkg.WaitFor(ctx, "wait for next instance check.", 2*time.Second)
	}
}
func (tikvRelOpts *TiKVRelOptions) getNormalTiKVInstance(instanceIdList []int, falutInstanceIdList []int) []int {
	var normalTiKVInstance []int
	instanceLen := len(falutInstanceIdList)
	if instanceLen == 1 {
		for _, instanceId := range instanceIdList {
			if instanceId != falutInstanceIdList[0] {
				normalTiKVInstance = append(normalTiKVInstance, instanceId)
			}
		}
	} else {
		normalTiKVInstance = instanceIdList[2:]
	}
	return normalTiKVInstance
}

func (tikvRelOpts *TiKVRelOptions) tikvOomCheck(ctx context.Context, tc resource.TiDBCluster, g *errgroup.Group, runTime time.Duration) {
	if tikvRelOpts.FaultDuration > 2*time.Hour {
		g.Go(WithGinkgoRecover(func() {
			log.Info("TiKV oom check.")
			oltppkg.InstanceOomCheck(ctx, tc, runTime, resource.TiKV)
			log.Info("TiKV oom check done.")
		}))
	}
}

func (tikvRelOpts *TiKVRelOptions) getOomCheckParams() oltppkg.OomCheckParameters {
	var allComponentOomCheck bool
	var tikvOomCheck bool
	if tikvRelOpts.FaultDuration > 2*time.Hour {
		allComponentOomCheck = false
		tikvOomCheck = false
	} else {
		allComponentOomCheck = true
		tikvOomCheck = true
	}
	oomCheckParams := oltppkg.OomCheckParameters{
		AllComponentOomCheck: allComponentOomCheck,
		TiKVOomCheck:         tikvOomCheck,
		PDOomCheck:           true,
		TiDBOomCheck:         true,
	}
	return oomCheckParams
}

func (tikvRelOpts *TiKVRelOptions) getRestartCheckParams() oltppkg.RestartCheckParameters {
	var allComponentRestartCheck bool
	var tikvRestartCheck bool
	if tikvRelOpts.FaultType == "failure" || tikvRelOpts.FaultType == "restart" {
		allComponentRestartCheck = false
		tikvRestartCheck = false
	} else {
		allComponentRestartCheck = true
		tikvRestartCheck = true
	}
	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: allComponentRestartCheck,
		TiKVRestartCheck:         tikvRestartCheck,
		PDRestartCheck:           true,
		TiDBRestartCheck:         true,
	}
	return restartCheckParms
}

func (tikvRelOpts *TiKVRelOptions) getRaftEngineGcCheckParams() oltppkg.RaftEngineAppendFileCountParameters {
	var raftEngineAppendFileCountTolerateDur time.Duration
	var raftEngineAppendFileCountIsCheck bool
	if tikvRelOpts.FaultDuration > 30*time.Minute {
		raftEngineAppendFileCountIsCheck = false
	} else {
		raftEngineAppendFileCountIsCheck = true
		raftEngineAppendFileCountTolerateDur = 300 * time.Second
	}
	raftEngineAppendFileCountCheckParms := oltppkg.RaftEngineAppendFileCountParameters{
		IsCheck:       raftEngineAppendFileCountIsCheck,
		TolerateCount: 200,
		TolerateDur:   raftEngineAppendFileCountTolerateDur,
	}
	return raftEngineAppendFileCountCheckParms
}

func (tikvRelOpts *TiKVRelOptions) WaitClusterStatusRecover(ctx context.Context, tc resource.MySQLCompatible, waitForRaftLogCatchUpTime time.Duration, waitForBalanceTime time.Duration) {
	oltppkg.LogAndBy("Waiting for raft log lag to be less than a certain value after recover fault.")
	// known incall https://internal.pingcap.net/jira/browse/INCALL-134
	oltppkg.WaitForRaftLogLagLessThan(ctx, tc, waitForRaftLogCatchUpTime, 2*time.Minute, 1000)

	oltppkg.LogAndBy("Waiting for raft engine append file count to be less than a certain value after recover fault.")
	oltppkg.WaitForRaftEngineAppendFileCountLessThan(ctx, tc, 120*time.Minute, 2*time.Minute, 100, 5*time.Minute)

	oltppkg.LogAndBy("Waiting for balance.")
	oltppkg.WaitForBalance(ctx, tc, waitForBalanceTime, 2*time.Minute, 0.3)
}