package oltp_tiup_test

import (
	"context"
	"database/sql"
	"flag"
	"fmt"
	"math/rand"
	"net/url"
	"os"
	"strconv"
	"strings"
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"golang.org/x/sync/errgroup"

	"github.com/pingcap/endless/pkg/client/tidb"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/pkg/workloadnode"
	"github.com/pingcap/endless/pkg/workloadnode/tiup"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/endless/testcase/oltp/systemtest/ddl"
	"github.com/pingcap/endless/testcase/oltp/systemtest/pdfun"
	"github.com/pingcap/endless/testcase/oltp/systemtest/query"
	"github.com/pingcap/endless/testcase/oltp/systemtest/scale"
	"github.com/pingcap/endless/testcase/oltp/systemtest/workload"
	"github.com/pingcap/endless/testcase/oltp/tiuptest/drautosync"
	"github.com/pingcap/endless/testcase/oltp/tiuptest/tiuppdrel"
	"github.com/pingcap/endless/testcase/oltp/tiuptest/tiuppkg"
	"github.com/pingcap/endless/testcase/oltp/tiuptest/tiupscale"
	"github.com/pingcap/endless/testcase/oltp/tiuptest/tiuptikvrel"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
	"github.com/spf13/viper"
)

const (
	envPrefix         = "TI_"
	envResParamPrefix = envPrefix + "PARAM_RES_"
)

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
		// tiupVersion string
		// tiupMirror  string
	)

	// flagSet.StringVar(&tiupVersion, "version", "nightly", "")
	// flagSet.StringVar(&tiupMirror, "tiupMirror", "staging", "")

	//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("Parser config files")
		ParseCaseParms(configFile, caseCmdParmsStr, &caseCmdParms)
		caseParams.CaseCmdParms = caseCmdParms

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

		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)
		pdfun.TestSuiteCommon(&caseParams)
		tiuppdrel.TestSuite(&caseParams)
		tiuptikvrel.TestSuite(&caseParams)
		tiupscale.TestSuite(&caseParams)
		drautosync.TestSuite(&caseParams)
	})
	AfterEach(func() {
		defer func() {
			if tc, ok := caseParams.Tc.(*tiup.TiUPCluster); ok {
				Try(tiuppkg.UploadClinic(tc, "--include", "monitor,log", "-f", "-3h", "-R", "tikv,pd"))
			}
		}()
		oltppkg.LogAndBy("End One Test")
		select {
		case <-caseParams.SkipAfterEachCheck:
			log.Info("Skip after each check.")
			return
		default:
		}
		if caseCmdParms.TiupClusterConfig.Topo == "dr-auto-sync" {
			log.Info("No balance check for dr auto sync tests")
			return
		}

		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, caseParms *oltppkg.CaseParams) {
	Try(os.Setenv("ENDLESS_CMD_STDOUT", "stdout"))
	caseParms.Ctx = ctx
	_, ok := os.LookupEnv(envResParamPrefix + "tc")
	if !ok {
		tiupNode := ctx.Resource("tiup").(resource.WorkloadNode)
		tiupConfig := caseParms.CaseCmdParms.TiupClusterConfig
		var tiupTopo string
		if tiupConfig.Topo == "dr-auto-sync" {
			if tiupConfig.TiupDrAutoSyncConfig.ReplicaNum == 5 {
				tiupTopo = fmt.Sprintf(DrAutoSyncTopoTmp, 5, 3, 2)
			} else {
				tiupTopo = fmt.Sprintf(DrAutoSyncTopoTmp, 3, 2, 1)
			}
		} else {
			tiupTopo = fmt.Sprintf(CommonTopoTmp, strconv.FormatBool(caseParms.CaseCmdParms.TiupClusterConfig.EnableTls))
		}
		tc := tiuppkg.InitTiupCluster(ctx, tiupNode, tiupTopo, caseParms.CaseCmdParms.TiupClusterConfig.ClusterName, caseParms.CaseCmdParms.TiupClusterConfig.Version, tiupConfig.TiupMirror, false)

		if tiupConfig.Topo == "dr-auto-sync" && !tiupConfig.EnableTls {
			if oltppkg.GetRepliactionStatus(&tc) != "sync" {
				drautosync.DestroyPrimaryCluster(ctx, tiupConfig.ClusterName, tiupNode)
				tc = tiuppkg.InitTiupCluster(ctx, tiupNode, tiupTopo, caseParms.CaseCmdParms.TiupClusterConfig.ClusterName, caseParms.CaseCmdParms.TiupClusterConfig.Version, tiupConfig.TiupMirror, false)
			}
			if tiupConfig.TiupDrAutoSyncConfig.PrimayBackupTikvNum == "6:6" {
				Try(tc.ScaleOut(drautosync.DrAutoSyncTopoScaleOut))
				promURL := Try(tc.ServiceURL(resource.Prometheus)).(*url.URL)
				oltppkg.BasicClusterExecPdCtl(&tc, "store limit all 200")
				timeOut := 10 * time.Minute
				oltppkg.WaitFor(ctx, "waiting for balance", 2*time.Minute)
				scale.WaitForOnline(promURL, 0, 1*time.Minute, timeOut)
			}
			caseParms.Tc = &tc
			drautosync.SetDrAutoSyncPlacementRule(ctx, tiupNode, caseParms)
			drautosync.WaitForReplicateStatus(&tc, 30*time.Minute, 1*time.Minute, "sync")
		}
		Try(tiuppkg.SetBlockCacheInRedhat(&tc))
		caseParms.Tc = &tc

	} else {
		caseParms.Tc = ctx.Resource("tc").(resource.TiDBCluster)
	}
	caseParms.Br = ctx.Resource("benchtoolset").(resource.BR)
	caseParms.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, caseParms oltppkg.CaseParams, caseCmdParms oltppkg.CaseCmdParms, s3Parms oltppkg.S3Parms) {
	oltppkg.StopWorkload(caseParms.BenchToolset, "goTpc")
	oltppkg.StopWorkload(caseParms.BenchToolset, "sysBench")
	oltppkg.WaitFor(ctx, "Wait 1m for cluster ready", 60*time.Second)

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

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

	log.Info("Check if the environment is normal.")
	oltppkg.CheckClusterHealth(caseParms.Tc)
	Try(oltppkg.BasicClusterExecPdCtl(caseParms.Tc, "scheduler remove evict-leader-scheduler"))

	oltppkg.LogAndBy("Import data")
	_, ok := caseParms.Tc.(resource.TiDBCluster)
	if ok {
		oltppkg.BrRestoreData(ctx, caseParms.Tc, caseParms.Br, s3Parms, caseCmdParms)
	} else {
		if strings.HasPrefix(caseCmdParms.WorkloadType.Type[0], "tpcc") {
			if caseCmdParms.TpccParms.Warehouse == 10 {
				TpccPrepare(ctx, caseParms.Tc.(*tiup.TiUPCluster), caseCmdParms.TpccParms.Db, caseParms.BenchToolset)
			} else {
				storageUrl := caseCmdParms.TpccParms.Storage
				tidbCli := tidb.NewClient(Try(caseParms.Tc.GetDB("")).(*sql.DB))
				if !Try(tidbCli.IsTableExist(ctx, "tpcc", "new_order")).(bool) {
					tiuppkg.TiUPRestoreWithBR(ctx, caseParms.Tc.(*tiup.TiUPCluster), storageUrl)
				}
			}
		}
		if strings.HasPrefix(caseCmdParms.WorkloadType.Type[0], "sysbench") {
			storageUrl := caseCmdParms.SysbenchParms.Storage
			tidbCli := tidb.NewClient(Try(caseParms.Tc.GetDB("")).(*sql.DB))
			if !Try(tidbCli.IsTableExist(ctx, "test", "sbtest1")).(bool) {
				tiuppkg.TiUPRestoreWithBR(ctx, caseParms.Tc.(*tiup.TiUPCluster), storageUrl)
			}
		}
	}
	if !oltppkg.CheckIfHeterogeneousDisk(caseParms.Tc) && caseCmdParms.TiupClusterConfig.Topo != "dr-auto-sync" {
		log.Info("Waiting for cluster balance.")
		Try(oltppkg.BasicClusterExecPdCtl(caseParms.Tc, "store limit all 200"))
		oltppkg.WaitForBalance(context.Background(), caseParms.Tc, 3*time.Hour, 5*time.Minute, 0.5)
		Try(oltppkg.BasicClusterExecPdCtl(caseParms.Tc, "store limit all 15"))
	}
}

// func getGinkgoTestName() string {
// 	spec := CurrentSpecReport()
// 	testName := spec.FullText()
// 	if testName == "" {
// 		panic("current test name is empty")
// 	}
// 	return testName
// }
func TpccPrepare(ctx context.Context, tc *tiup.TiUPCluster, dbName string, benchToolset resource.WorkloadNode) {
	Try(oltppkg.RecreateDatabase(ctx, tc, dbName))
	dbURL := Try(tc.ServiceURL(resource.DBAddr)).(*url.URL)
	tpcc := oltppkg.NewTpcc("prepare")
	tpcccmd := tpcc.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Warehouses(10).Threads(10).Time("36000m")
	runcmd := tpcccmd.Build()
	cmd := workloadnode.CommandContext(ctx, runcmd)
	Try(cmd.Run(benchToolset))
}