package tikvrel

import (
	"context"
	"errors"
	"io"
	"math/rand"
	"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"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"
	"golang.org/x/sync/errgroup"
)

func OLTP_STABILITY_REL_002(
	ctx context.Context,
	tc resource.TiDBCluster,
	pr *io.PipeReader,
	caseCmdParms oltppkg.CaseCmdParms,
	skipAfterEachCheck chan bool,
) {
	oltppkg.LogAndBy("Randomly schedule and inject tikv network fault")
	skipAfterEachCheck <- true
	instanceType := resource.TiKV
	warmUpTime := 10 * time.Minute
	faultDuration := oltppkg.GetRandDuration(time.Minute, 5) + 5*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)
	tikvFaultTpyes := []string{"network-loss", "network-delay"}
	rand.Seed(time.Now().UnixNano())
	faultTpye := tikvFaultTpyes[rand.Intn(len(tikvFaultTpyes))]
	var c chaos.Chaos
	switch faultTpye {
	case "network-loss":
		c.ExperimentSpec = &resource.NetworkLossSpec{
			Loss:        oltppkg.GetRandIntStr(50, 100),
			Correlation: "25",
		}
	case "network-delay":
		c.ExperimentSpec = &resource.NetworkLatencySpec{
			Duration:    oltppkg.GetRandIntStr(1, 5) + "s",
			Scheduler:   &resource.ExperimentScheduler{Cron: "@every 5s"},
			Latency:     "1s",
			Correlation: "25",
		}
	case "IODelay":
		c.ExperimentSpec = &resource.IODelaySpec{
			VolumePath: "/var/lib/tikv",
			Path:       "/var/lib/tikv/data/**/*",
			Delay:      oltppkg.GetRandIntStr(100, 500) + "ms", //base delay: 40ms
			Percent:    100,
		}
	default:
		err := errors.New("invalid faultTpye")
		Expect(err).Should(Succeed())
	}
	c = Try(oltppkg.InitChaos(tc, instanceType, faultTpye, c.ExperimentSpec, &instanceIdList[instanceId])).(chaos.Chaos)
	//Error 8027: https://github.com/pingcap/tidb/issues/30258
	//Error 9001: PD server timeout (Turning on random scheduling is easy to appear)
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   false,
		IgnoreErr: append(strings.Split(caseCmdParms.TpccParms.IgnoreErrors, ","), "9001", "invalid connection", "err dial tcp"),
	}

	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck: false,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck: false,
	}

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

	pd := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	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)

	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())
			oltppkg.WaitFor(gctx, "Recovery after inject fault", recoveryTime)
		}
		oltppkg.LogAndBy("Waiting for balance.")
		oltppkg.WaitForBalance(gctx, tc, waitForBalanceTime, 5*time.Minute, 0.3)
		cancel()
	}))
	Expect(g.Wait()).Should(BeNil())
}