package datarecovery

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

	_ "github.com/go-sql-driver/mysql"
	. "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/log"

	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/test-infra/sdk/resource"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"
	"golang.org/x/sync/errgroup"
)

func OLTP_DATA_ONLINE_RECOVERY_FUN_001(
	ctx context.Context,
	tc resource.TiDBCluster,
	benchToolset resource.WorkloadNode,
	workloadRunParms oltppkg.WorkloadRunParms,
	caseCmdParms oltppkg.CaseCmdParms,
) {
	tikvVer := oltppkg.GetCompVersion(ctx, tc, "tikv")
	curVerShort := strings.TrimSuffix(strings.TrimPrefix(tikvVer, "v"), "-pre")
	if oltppkg.CompareVersion(curVerShort, "6.1.0") < 0 {
		log.Info(fmt.Sprintf("tikv version %s do not support this feature.", curVerShort))
		return
	}
	oltppkg.LogAndBy("Test for Online Recovery When Raft Majority Fail")
	pd := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	instanceType := resource.TiKV
	// instanceIdList := GetAllTiKVActiveInstanceId(tc)
	instanceIdList := oltppkg.GetInstanceIdList(tc, instanceType)
	experimentSpec := &resource.FailureExperimentSpec{}
	c1 := Try(oltppkg.InitChaos(tc, instanceType, "failure", experimentSpec, &instanceIdList[0])).(chaos.Chaos)
	c2 := Try(oltppkg.InitChaos(tc, instanceType, "failure", experimentSpec, &instanceIdList[1])).(chaos.Chaos)
	node1 := Try(tc.GetNodes(instanceType)).([]resource.TiDBComponentNode)[0]
	node2 := Try(tc.GetNodes(instanceType)).([]resource.TiDBComponentNode)[1]
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   false,
		IgnoreErr: append(strings.Split(caseCmdParms.TpccParms.IgnoreErrors, ","), "9002", "9005"),
	}
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck: false,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck: false,
	}
	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: false,
		TiKVRestartCheck:         false,
		PDRestartCheck:           true,
		TiDBRestartCheck:         true,
	}
	workloadRunParms.InitAndStart(ctx, tc, benchToolset, caseCmdParms, false)
	warmUpTime := 30 * time.Minute
	workloadRunAfterFault := 5 * time.Minute
	runTime := warmUpTime + workloadRunAfterFault - 1*time.Minute
	ctxAll, cancel := context.WithTimeout(ctx, runTime)
	defer cancel()
	g, gctx := errgroup.WithContext(ctxAll)
	oltppkg.FunCommonCheck(gctx, tc, g, runTime, workloadRunParms.Cmdpr, qpsJitterCheckParms, durJitterCheckParms, restartCheckParms, workloadCheckParms)
	oltppkg.WaitFor(ctx, "wait for down two tikv", warmUpTime)
	oltppkg.LogAndBy("step1: down two tikv")
	Expect(c1.Run()).Should(Succeed())
	Expect(c2.Run()).Should(Succeed())
	defer func() {
		Try(pd.SetAllStoreLimit(15))
		Expect(c1.Clean()).Should(Succeed())
		Expect(c2.Clean()).Should(Succeed())
	}()
	oltppkg.WaitFor(ctx, "wait for stop workload", workloadRunAfterFault)
	oltppkg.StopWorkload(workloadRunParms.WorkloadNode, "goTpc")
	oltppkg.StopWorkload(workloadRunParms.WorkloadNode, "sysBench")
	Expect(g.Wait()).Should(BeNil())

	oltppkg.LogAndBy("step2: execute online unsafe recovery")
	failedStoresId := GetFailedStoresId(tc)
	Expect(len(failedStoresId)).Should(Equal(2))
	Try(ExecuteOnlineUnsafeRecovery(tc, failedStoresId))

	oltppkg.LogAndBy("step3: wait for recovery success")
	Try(GetOnlineRecoveryResult(ctx, tc, 2*time.Minute, 20*time.Minute))
	Expect(GetTombstoneNum(tc)).Should(Equal(float64(2)))

	oltppkg.WaitFor(ctx, "wait for run workload", 2*time.Minute)
	oltppkg.LogAndBy("step4: run workload after recovery success")
	runTime = 5 * time.Hour
	ctxAll, cancel = context.WithTimeout(ctx, runTime)
	defer cancel()
	g, gctx = errgroup.WithContext(ctxAll)
	Try(pd.SetAllStoreLimit(200))
	workloadRunParms.InitAndStart(ctx, tc, benchToolset, caseCmdParms, false)
	oltppkg.FunCommonCheck(gctx, tc, g, runTime, workloadRunParms.Cmdpr, workloadCheckParms)
	//1、kubectl -n ${namespace} get pvc -l tidb.pingcap.com/pod-name=${pod_name}
	//2、kubectl delete -n ${namespace} pvc ${pvc_name} --wait=false
	//3、kubectl delete -n ${namespace} pod ${pod_name}
	node1.CleanData()
	node2.CleanData()
	// https://github.com/tikv/pd/issues/5093
	// WaitForStoreRegionsToZero(ctx, tc, failedStoresId, 5*time.Minute, runTime)
	// RemoveTombstone(ctx, tc, 1*time.Minute, 10*time.Minute)
	oltppkg.LogAndBy("Waiting for balance.")
	oltppkg.WaitForBalance(ctx, tc, 120*time.Minute, 2*time.Minute, 0.3)
	cancel()
}