package rawkv_scenario_test

import (
	"context"
	"flag"
	"fmt"
	"strings"

	. "github.com/onsi/ginkgo"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"github.com/prometheus/client_golang/api"
	v1 "github.com/prometheus/client_golang/api/prometheus/v1"
	"go.uber.org/zap"

	"github.com/pingcap/endless/pkg/workloadnode/tiup"

	. "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/rawkv/scenario/ccb_application_router"

	_ "github.com/pingcap/endless/pkg/client/tidb"
)

var _ = ParameterizedDescribe("rawkv", func(flagSet *flag.FlagSet) {
	var (
		// Workload related vars.
		IsTiUPCluster              bool
		Version                    string
		APIVersion                 int
		ReplicaCount               int
		TiKVCount                  int
		TiDBCount                  int
		PDCount                    int
		ZoneCount                  int
		WorkloadNodeCount          int
		instancesCount             ccb_application_router.InstancesCount
		testbedName                string
		RestorePath                string
		clientMonitorAPIPrometheus v1.API
		// Cluster created by operator.
		tc resource.TiDBCluster
		// Cluster created by tiup
		tiupWorkloadNode resource.WorkloadNode
		tiupCluster      *tiup.TiUPCluster
		DeployUser       string
		// tiupClusterAPIPrometheus v1.API

		// Test case control parameters.
		// When runTime is reached, the ctx becomes timeout.
		rootCtx    context.Context
		rootCancel context.CancelFunc
	)
	// Workload flags.
	flagSet.BoolVar(&IsTiUPCluster, "IsTiUPCluster", false, "The cluster is created by tiup or not")
	flagSet.StringVar(&Version, "version", "v5.0.4", "Verion of Cluster")
	flagSet.IntVar(&APIVersion, "APIVersion", 2, "API Verion")
	flagSet.IntVar(&ReplicaCount, "MaxReplica", 5, "Max Replicas")
	flagSet.StringVar(&DeployUser, "DeployUser", "tidb", "Deploy User specified")
	flagSet.IntVar(&TiKVCount, "TiKVCount", 4, "the total number of tikv (There are 4 zones/AZs)")
	flagSet.IntVar(&TiDBCount, "TiDBCount", 0, "the total number of tidb nodes")
	flagSet.IntVar(&PDCount, "PDCount", 3, "the total number of PD")
	flagSet.IntVar(&ZoneCount, "ZoneCount", 4, "the total number of Zones, should not be larger than 4")
	flagSet.IntVar(&WorkloadNodeCount, "WorkloadNodeCount", 2, "the total number of workload nodes")
	flagSet.StringVar(&RestorePath, "RestorePath", "", "Backuped Data on minio")

	BeforeEach(func() {
		// Initialize config.
		log.Info("BeforeEach")
		log.Info("IsTiUPCluster:", zap.Bool("IsTiUPCluster", IsTiUPCluster))
		log.Info("TiUPCluster Version:", zap.String("Version", Version))
		// Initialize root context.
		rootCtx, rootCancel = context.WithCancel(suiteTestCtx)
		instancesCount = ccb_application_router.InstancesCount{
			PDCount:           PDCount,
			TiDBCount:         TiDBCount,
			TiKVCount:         TiKVCount,
			ZoneCount:         ZoneCount,
			WorkloadNodeCount: WorkloadNodeCount,
			ReplicaCount:      ReplicaCount,
		}
		// Initialize resources.
		if IsTiUPCluster {
			tiupWorkloadNode = suiteTestCtx.Resource("tiup").(resource.WorkloadNode)
			PDTopo := ccb_application_router.GetRawKVClusterPDTopo(PDCount)
			TiDBTopo := ccb_application_router.GetRawKVClusterTiDBTopo(TiDBCount, Version, APIVersion)
			TiKVTopo := ccb_application_router.GetRawKVClusterTiKVTopo(TiKVCount, ZoneCount)
			tiupTopo := fmt.Sprintf(ccb_application_router.TiupTopoTemplate, DeployUser, ccb_application_router.DeployDir, ccb_application_router.DataDir, ReplicaCount, ccb_application_router.GetStorageAPIVersion(APIVersion, Version), PDTopo, TiDBTopo, TiKVTopo)
			tiupCluster = tiup.NewTiUPCluster(tiupWorkloadNode, "tiup-peer", tiupTopo, "rawkv")
			testbedName = tiupCluster.TestBed
			// avoid failling to download PD, TiKV images.
			cmd := fmt.Sprintf("tiup mirror set %s", "http://172.16.5.134:8987")
			// set staging.tiup-server.pingcap.net
			options := resource.WorkloadNodeExecOptions{Command: cmd}
			_, _, _, err := tiupWorkloadNode.Exec(options)
			Expect(err).Should(BeNil())
			cmd = fmt.Sprintf("tiup mirror set %s", "http://172.16.5.139:8988")
			options = resource.WorkloadNodeExecOptions{Command: cmd}
			_, _, _, err = tiupWorkloadNode.Exec(options)
			Expect(err).Should(BeNil())
			err = tiupCluster.Deploy(Version)
			if err != nil {
				if strings.Contains(err.Error(), "Cluster name 'rawkv' is duplicated") {
					log.Info("Cluster name 'rawkv' has already deployed")
				} else {
					Expect(err).Should(BeNil())
				}
			} else {
				Try(tiupCluster.Start())
				if Version == "v5.0.4" {
					log.S().Info("Patch v5.0.4")
					options := resource.WorkloadNodeExecOptions{Command: "tiup cluster patch -y rawkv tikv-server-v5.0.4.tar.gz -R tikv  --overwrite"}
					_, _, _, err = tiupWorkloadNode.Exec(options)
					Expect(err).Should(BeNil())
				}
			}
			log.Info("Update PD config")
			placementRule := ccb_application_router.GeneratePlacementRules(TiKVCount, ZoneCount)
			ccb_application_router.SetupPDWithTiUP(tiupCluster, tiupWorkloadNode, placementRule, rootCtx)
			log.Info("Update client info into ccbload clientmonitor Prometheus Config")
			ccb_application_router.UpdatePrometheusConfig(suiteTestCtx, rootCtx, WorkloadNodeCount)
		} else {
			tc = suiteTestCtx.Resource("tc").(resource.TiDBCluster)
			log.Info("Check the cluster health status before running the case")
			oltppkg.CheckClusterHealth(tc)
			testbedName = tc.Handle().Testbed
			// Todo: update PD config and config prometheus to scrape workloads metrics
		}
		log.Info("Testbed Name:", zap.String("testbedName", testbedName))
		// TODO: config flag param if the test case is defined clientmonitor prometheus
		ccbloadClientPrometheus, err := api.NewClient(api.Config{
			Address: fmt.Sprintf("http://clientmonitor-0.clientmonitor-peer.%s:9090", testbedName),
		})
		Expect(err).Should(BeNil())
		clientMonitorAPIPrometheus = v1.NewAPI(ccbloadClientPrometheus)
	})

	AfterEach(func() {
		log.Info("AfterEach")
		if IsTiUPCluster {
			url, err := tiupCluster.Diag().WithUploadMaxSize(1024 * 1024 * 1024).CollectAndUpload(tiup.DiagCollectAndUploadOptions{
				Collect: tiup.DiagCollectOptions{
					Flags: []string{"--include", "monitor,log"}},
				Upload: tiup.DiagUploadOptions{Token: "eyJrIjoiS3hFbHZQVGRIUFBuTlBCTCIsInUiOjk5LCJpZCI6NzN9"}})
			log.S().Info("clinic url: with err info: %v", url, err)
		}
		rootCancel()
	})

	Context("scenario", func() {
		TestCase("ccb_app_router_launch", "ccb_scenario_001", func() {
			if IsTiUPCluster {
				ccb_application_router.CCBLaunchWithImportDataByTiUP(suiteTestCtx, rootCtx, tiupCluster, clientMonitorAPIPrometheus, instancesCount, APIVersion)
			} else {
				ccb_application_router.CCBLaunchWithImportData(suiteTestCtx, rootCtx, tc, clientMonitorAPIPrometheus, APIVersion)
			}
		})
		TestCase("ccb_app_router_launch", "ccb_scenario_002", func() {
			if IsTiUPCluster {
				ccb_application_router.CCBLaunchWithRestoreDataByTiUP(suiteTestCtx, rootCtx, tiupCluster, clientMonitorAPIPrometheus, instancesCount, APIVersion)
			} else {
				PathOnMinio := fmt.Sprintf("s3://%s?access-key=minioadmin&secret-access-key=minioadmin&endpoint=http://minio.pingcap.net:9000&force-path-style=true", RestorePath)
				ccb_application_router.CCBLaunchWithRestoreData(suiteTestCtx, rootCtx, tc, clientMonitorAPIPrometheus, APIVersion, PathOnMinio)
			}
		})
		TestCase("ccb_app_router_launch", "ccb_scenario_003", func() {
			ccb_application_router.CCBLaunchWithBatchPutData(suiteTestCtx, rootCtx, tc, clientMonitorAPIPrometheus, APIVersion)
		})
		TestCase("ccb_app_router_ha_test", "ccb_scenario_004", func() {
			PathOnMinio := fmt.Sprintf("s3://%s?access-key=minioadmin&secret-access-key=minioadmin&endpoint=http://minio.pingcap.net:9000&force-path-style=true", RestorePath)
			ccb_application_router.HATestWithCCBWorkloadByTiUP(suiteTestCtx, rootCtx, tiupCluster, clientMonitorAPIPrometheus, instancesCount, APIVersion, PathOnMinio)
		})
		TestCase("ccb_app_router_api_v2_online_test", "ccb_scenario_005", func() {
			// TODO: maybe one day test-infra could get zone info from rawkv cluster created by operator
			ccb_application_router.RunAPIV2OnlineProcedure(suiteTestCtx, rootCtx, tiupCluster, clientMonitorAPIPrometheus, instancesCount)
		})
		TestCase("ccb_app_router_api_v2_100e_data_br_test", "ccb_scenario_006", func() {
			// TODO: maybe one day test-infra could get zone info from rawkv cluster created by operator
			ccb_application_router.RunAPIV2BR100eData(suiteTestCtx, rootCtx, tiupCluster, clientMonitorAPIPrometheus, instancesCount, APIVersion)
		})
		TestCase("ccb_app_router_recover_disaster_test", "ccb_scenario_007", func() {
			// As test-infra does not support attaining zone info of rawkv cluster created by operator
			// so ccb_scenario_002 runs on cluster created by tiup
			// TODO: maybe one day test-infra could get zone info from rawkv cluster created by operator
			PathOnMinio := fmt.Sprintf("s3://%s?access-key=minioadmin&secret-access-key=minioadmin&endpoint=http://minio.pingcap.net:9000&force-path-style=true", RestorePath)
			ccb_application_router.DisasterTestWithCCBWorkloadByTiUP(suiteTestCtx, rootCtx, tiupCluster, clientMonitorAPIPrometheus, instancesCount, APIVersion, PathOnMinio)

		})
	})
})