package oltp_test

import (
	"context"
	"flag"
	"fmt"
	"math/rand"
	"os"
	"strings"
	"time"

	. "github.com/onsi/ginkgo"

	. "github.com/onsi/gomega"

	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/endless/testcase/oltp/systemtest/availzone"
	"github.com/pingcap/endless/testcase/oltp/systemtest/configupdate"
	"github.com/pingcap/endless/testcase/oltp/systemtest/datarecovery"
	"github.com/pingcap/endless/testcase/oltp/systemtest/ddl"
	"github.com/pingcap/endless/testcase/oltp/systemtest/gc"
	"github.com/pingcap/endless/testcase/oltp/systemtest/loadsplit"
	"github.com/pingcap/endless/testcase/oltp/systemtest/pdfun"
	"github.com/pingcap/endless/testcase/oltp/systemtest/pdrel"
	"github.com/pingcap/endless/testcase/oltp/systemtest/profiling"
	"github.com/pingcap/endless/testcase/oltp/systemtest/query"
	"github.com/pingcap/endless/testcase/oltp/systemtest/scale"
	"github.com/pingcap/endless/testcase/oltp/systemtest/tidbrel"
	"github.com/pingcap/endless/testcase/oltp/systemtest/tikvrel"
	"github.com/pingcap/endless/testcase/oltp/systemtest/upgrade"
	"github.com/pingcap/endless/testcase/oltp/systemtest/workload"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
	"github.com/spf13/viper"
	"golang.org/x/sync/errgroup"
)

var _ = ParameterizedDescribe("TiDB-OLTP", func(flagSet *flag.FlagSet) {
	var (
		ctx core.TestContext

		//command line parameter for workload
		s3Parms oltppkg.S3Parms
		//can use config file if debug locally
		configFile string

		//all tcms case cmd parameters
		caseCmdParmsStr string
		caseCmdParms    oltppkg.CaseCmdParms

		//parameter used by case
		caseParams oltppkg.CaseParams
	)
	//import data address
	flagSet.StringVar(&s3Parms.S3Endpoint, "s3-endpoint", "http://minio.pingcap.net:9000", "")
	flagSet.StringVar(&s3Parms.S3AccessKeyID, "s3-key", "minioadmin", "")
	flagSet.StringVar(&s3Parms.S3SecretAccessKey, "s3-secret", "minioadmin", "")

	// all tcms case cmd parameters
	flagSet.StringVar(&caseCmdParmsStr, "caseCmdParms", oltppkg.CaseCmdParmsDefault, "")

	// can use config file if debug locally
	flagSet.StringVar(&configFile, "configFile", "config.yaml", "")

	BeforeEach(func() {
		ctx = suiteTestCtx
		caseParams = oltppkg.CaseParams{}
		caseParams.S3Parms = s3Parms

		oltppkg.LogAndBy("Init resource")
		InitResource(ctx, &caseParams)

		oltppkg.LogAndBy("Parser config files")
		ParseCaseParms(configFile, caseCmdParmsStr, &caseCmdParms)
		caseParams.CaseCmdParms = caseCmdParms

		oltppkg.LogAndBy("Check cluster and import data")
		CheckAndPrepareData(ctx, caseParams, caseCmdParms, s3Parms)

		if !caseCmdParms.SkipRunWorkload {
			oltppkg.LogAndBy("Start workload")
			caseParams.WorkloadRunParms.InitAndStart(ctx, caseParams.Tc, caseParams.BenchToolset, caseCmdParms, true)
		}
		caseParams.SkipAfterEachCheck = make(chan bool, 1)
	})

	Context("run workload", func() {
		workload.TestSuite(&caseParams)
		ddl.TestSuite(&caseParams)
		query.TestSuite(&caseParams)
		scale.TestSuite(&caseParams)
		tikvrel.TestSuite(&caseParams)
		tidbrel.TestSuite(&caseParams)
		profiling.TestSuite(&caseParams)
		configupdate.TestSuite(&caseParams)
		datarecovery.TestSuite(&caseParams)
		upgrade.TestSuite(&caseParams)
		gc.TestSuite(&caseParams)
		pdrel.TestSuite(&caseParams)
		pdfun.TestSuiteTools(&caseParams)
		pdfun.TestSuiteCommon(&caseParams)
		pdfun.TestSuiteStability(&caseParams)
		pdfun.TestSuiteFeatures(&caseParams)
		availzone.TestSuite(&caseParams)
		loadsplit.TestSuite(&caseParams)
	})
	AfterEach(func() {
		oltppkg.LogAndBy("End One Test")
		select {
		case <-caseParams.SkipAfterEachCheck:
			log.Info("Skip after each check.")
			return
		default:
		}

		oltppkg.WaitFor(context.Background(), "waiting for workload running 2min after case finish", 2*time.Minute)
		oltppkg.LogAndBy("Post check 1: Test finished, check if all tikvs and pds are up.")
		oltppkg.CheckClusterHealth(caseParams.Tc)

		oltppkg.LogAndBy("Post check 2: Test finished, check if region size/leader/number if balance in all tikv instance.")
		ctx2, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
		defer cancel()
		g, ctxg := errgroup.WithContext(ctx2)
		threshhold := 0.3
		checkTime := 2 * time.Minute

		oltppkg.BalanceCheckAftWl(ctxg, caseParams.Tc, g, checkTime, threshhold)
		Expect(g.Wait()).Should(BeNil())
	})

})

func InitResource(ctx core.TestContext, clusterRes *oltppkg.CaseParams) {
	Try(os.Setenv("ENDLESS_CMD_STDOUT", "stdout"))
	clusterRes.Ctx = ctx
	clusterRes.Tc = ctx.Resource("tc").(resource.TiDBCluster)
	clusterRes.Br = ctx.Resource("br").(resource.BR)
	clusterRes.BenchToolset = ctx.Resource("benchtoolset").(resource.WorkloadNode)
}
func ParseCaseParms(configFile string, caseCmdParmsStr string, caseCmdParms *oltppkg.CaseCmdParms) {
	configFileSuffix := ".yaml"
	configFilePath := "."
	configFileName := strings.TrimSuffix(configFile, configFileSuffix)
	viper.SetConfigName(configFileName)
	viper.SetConfigType("yaml")
	viper.AddConfigPath(configFilePath)
	err := viper.ReadInConfig()
	if err == nil {
		log.Info(fmt.Sprintf("Use config file : %s\n", configFile))
		viper.Unmarshal(&caseCmdParms)
		log.Info(fmt.Sprintf("caseCmdParms: %v", caseCmdParms))
	} else {
		log.Info("The config file does not exist, use command line setting or oltppkg.CaseCmdParmsDefault.")
		*caseCmdParms = oltppkg.BuildCaseCmdParms(caseCmdParmsStr)
	}
	log.Info("Get workload list")
	if caseCmdParms.WorkloadType.Type[0] == "randomWorkload" {
		randomWorkloadScope := caseCmdParms.WorkloadType.RandomWorkloadScope
		randomWorkloadNum := len(randomWorkloadScope)
		rand.Seed(time.Now().UnixNano())
		workload := randomWorkloadScope[rand.Intn(randomWorkloadNum)]
		caseCmdParms.WorkloadType.Type = []string{workload}
	}
	log.Info(fmt.Sprintf("workloadTypeList %s", caseCmdParms.WorkloadType.Type))
}
func CheckAndPrepareData(ctx context.Context, clusterRes oltppkg.CaseParams, caseCmdParms oltppkg.CaseCmdParms, s3Parms oltppkg.S3Parms) {
	oltppkg.StopWorkload(clusterRes.BenchToolset, "goTpc")
	oltppkg.StopWorkload(clusterRes.BenchToolset, "sysBench")
	oltppkg.WaitFor(ctx, "Wait 20s to start workload to avoid db connection error after scale-in.", 20*time.Second)

	oltppkg.LogAndBy("Getting Cluster and BR Versions")
	oltppkg.ShowClusterVersions(clusterRes.Tc.(resource.TiDBCluster))
	oltppkg.ShowBrReleaseInfo(clusterRes.Br)

	oltppkg.LogAndBy("Setting Cluster Params")
	oltppkg.SetTidbVariables(ctx, clusterRes.Tc, caseCmdParms)

	log.Info("Check if the environment is normal.")
	oltppkg.CheckClusterHealth(clusterRes.Tc)
	Try(oltppkg.ExecPdCtl(clusterRes.Tc.(resource.TiDBCluster), "scheduler remove evict-leader-scheduler"))

	oltppkg.LogAndBy("Import data")
	oltppkg.BrRestoreData(ctx, clusterRes.Tc, clusterRes.Br, s3Parms, caseCmdParms)

	if !oltppkg.CheckIfHeterogeneousDisk(clusterRes.Tc) {
		log.Info("Waiting for cluster balance.")
		Try(oltppkg.ExecPdCtl(clusterRes.Tc.(resource.TiDBCluster), "store limit all 200"))
		oltppkg.WaitForBalance(context.Background(), clusterRes.Tc, 3*time.Hour, 5*time.Minute, 0.5)
		Try(oltppkg.ExecPdCtl(clusterRes.Tc.(resource.TiDBCluster), "store limit all 15"))
	}
}