package tikvrel

import (
	"context"
	"io"
	"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/matcher/metrics"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
)

func OLTP_STABILITY_REL_001(
	ctx context.Context,
	tc resource.TiDBCluster,
	pr *io.PipeReader,
	caseCmdParms oltppkg.CaseCmdParms,
	skipAfterEachCheck chan bool,
) {
	oltppkg.LogAndBy("Randomly schedule and inject tikv down for more than 30 minutes")
	oltppkg.ExecPdCtl(tc, "config set max-store-down-time 10m")
	defer oltppkg.ExecPdCtl(tc, "config set max-store-down-time 30m")
	skipAfterEachCheck <- true
	pd := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	instanceType := resource.TiKV
	warmUpTime := 10 * time.Minute
	faultDuration := 20 * time.Minute
	recoveryTime := 20 * time.Minute
	runTime := 20 * time.Hour
	faultRunTime := 10 * time.Hour
	waitForBalanceTime := runTime - faultRunTime
	instanceIdList := oltppkg.GetInstanceIdList(tc, instanceType)
	instanceId, _ := oltppkg.GetRandInstance(tc, instanceType)
	experimentSpec := &resource.FailureExperimentSpec{}
	c := Try(oltppkg.InitChaos(tc, instanceType, "failure", experimentSpec, &instanceIdList[instanceId])).(chaos.Chaos)

	//https://github.com/tikv/client-go/issues/382
	//Error 9001: PD server timeout (Turning on random scheduling is easy to appear)
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: append(strings.Split(caseCmdParms.TpccParms.IgnoreErrors, ","), "9001", "invalid connection", "err dial tcp"),
	}

	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck: false,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck: false,
	}
	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: false,
		TiKVRestartCheck:         false,
		PDRestartCheck:           true,
		TiDBRestartCheck:         true,
	}
	ctxAll, cancel := context.WithTimeout(ctx, runTime)
	defer cancel()
	g, gctx := errgroup.WithContext(ctxAll)

	Try(pd.SetAllStoreLimit(200))
	defer Try(pd.SetAllStoreLimit(15))
	oltppkg.SetPDRandomScheduler(tc, "add")
	defer oltppkg.SetPDRandomScheduler(tc, "remove")

	oltppkg.FunCommonCheck(gctx, tc, g, runTime, pr, qpsJitterCheckParms, durJitterCheckParms, workloadCheckParms, restartCheckParms)

	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(gctx, tc, dbName, 5*time.Minute)
	}))

	g.Go(WithGinkgoRecover(func() {
		oltppkg.WaitFor(gctx, "Wait for remove random scheduler", 6*time.Hour)
		oltppkg.SetPDRandomScheduler(tc, "remove")
	}))

	g.Go(WithGinkgoRecover(func() {
		oltppkg.WaitFor(gctx, "Warm Up before inject fault", warmUpTime)
		timeOut := time.After(faultRunTime)
	Loop:
		for {
			select {
			case <-gctx.Done():
				log.Info("inject fault ctx done.")
				return
			case <-timeOut:
				log.Info("inject fault test finish.")
				break Loop
			default:
			}
			err := oltppkg.InjectFaultByChaos(&c, gctx, faultDuration)
			Expect(err).Should(Succeed())
			log.Info("Recovery after inject fault", zap.Duration("recovery-time: ", recoveryTime))
			log.Info("TiKV restart check during recovery")
			ConsistentlyWithMetrics(tc, recoveryTime).
				ShouldNot(TiKVRestart().WithCtx(gctx))
			log.Info("TiKV restart check done during recovery.")
		}
		oltppkg.LogAndBy("Waiting for balance.")
		oltppkg.WaitForBalance(gctx, tc, waitForBalanceTime, 5*time.Minute, 0.5)
		cancel()
	}))
	Expect(g.Wait()).Should(BeNil())

	oltppkg.LogAndBy("Post check 1: Test finished, check if all tikvs and pds are up.")
	oltppkg.CheckClusterHealth(tc)
}