package tikvrel

import (
	"context"
	"io"
	"strings"
	"time"

	. "github.com/onsi/gomega"
	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"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"
	"golang.org/x/sync/errgroup"
)

func OLTP_STABILITY_REL_004(
	ctx context.Context,
	tc resource.TiDBCluster,
	pr *io.PipeReader,
	caseCmdParms oltppkg.CaseCmdParms,
) {
	oltppkg.LogAndBy("Randomly schedule and inject tikv down repeatedly")
	warmUpTime := 10 * time.Minute
	recoveryTime := 20 * time.Minute
	runTime := 15 * time.Hour
	faultRunTime := 10 * time.Hour
	waitForBalanceTime := runTime - faultRunTime
	//https://github.com/tikv/client-go/issues/382
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: append(strings.Split(caseCmdParms.TpccParms.IgnoreErrors, ","), "invalid connection", "err dial tcp"),
	}

	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: false,
		TiKVRestartCheck:         false,
		PDRestartCheck:           true,
		TiDBRestartCheck:         true,
	}

	ctxAll, cancel := context.WithTimeout(ctx, runTime)
	defer cancel()
	g, gctx := errgroup.WithContext(ctxAll)

	pd := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)

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

	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, "warm up before randomly restart instance", warmUpTime)
		timeOut := time.After(faultRunTime)
	Loop:
		for {
			select {
			case <-gctx.Done():
				log.Info("randomly restart ctx done.")
				return
			case <-timeOut:
				log.Info("inject fault test finish.")
				break Loop
			default:
			}
			oltppkg.RandomlyRestartInstance(tc, resource.TiKV)
			oltppkg.WaitFor(gctx, "wait for next TiKV restart", recoveryTime)
		}
		Try(pd.SetAllStoreLimit(200))
		defer Try(pd.SetAllStoreLimit(15))
		oltppkg.LogAndBy("Waiting for balance.")
		oltppkg.WaitForBalance(gctx, tc, waitForBalanceTime, 5*time.Minute, 0.3)
		cancel()
	}))
	Expect(g.Wait()).Should(BeNil())
}