package tiup

import (
	"context"
	"fmt"
	"strconv"
	"time"

	. "github.com/onsi/ginkgo"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"

	"github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/pkg/workloadnode/tiup"
	"github.com/pingcap/endless/testcase/tiflash/pkg"
	"github.com/pingcap/endless/testcase/tiflash/tiup/workload/ch"

	"github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/pkg/workloadnode"
)

// TODO
// 1. abstract workload for tiup and tidb operator
// 2. add tiup basic deploy and operator test
var _ = Describe("TiFlash", func() {
	var (
		version         string
		arch            string
		enableTls       string
		tiupMirror      string
		tidbNum         int
		tikvNum         int
		tiflashNum      int
		duration        time.Duration
		tiupMirrorSetUp string
		tc              *tiup.TiUPCluster
		wl              *ch.ChWorkload
		controlNode     resource.WorkloadNode
	)
	BeforeEach(func() {
		version = suiteTestCtx.Param("version", "nightly")
		arch = suiteTestCtx.Param("arch", "amd64")
		enableTls = suiteTestCtx.Param("enable_tls", "false")
		tiupMirror = suiteTestCtx.Param("mirror", "product")
		tidbNum = matcher.Try(strconv.Atoi(suiteTestCtx.Param("tidb_num", "2"))).(int)
		tikvNum = matcher.Try(strconv.Atoi(suiteTestCtx.Param("tikv_num", "3"))).(int)
		tiflashNum = matcher.Try(strconv.Atoi(suiteTestCtx.Param("tiflash_num", "2"))).(int)
		duration = matcher.Try(time.ParseDuration(suiteTestCtx.Param("duration", "3h"))).(time.Duration)

		tiupMirrorSetUp = fmt.Sprintf("tiup mirror set %s", productionMirror)
		if tiupMirror == "staging" {
			tiupMirrorSetUp = fmt.Sprintf("tiup mirror set %s", stagingMirror)
		}

		controlNodeName := "tiup" // Node 的 hostname = $name-peer
		controlNode = suiteTestCtx.Resource(controlNodeName).(resource.WorkloadNode)
		log.Info(tiupMirrorSetUp)
		setCmd := workloadnode.CommandContext(suiteTestCtx, tiupMirrorSetUp)
		matcher.Try(setCmd.Run(controlNode))
		log.Info("Mirror setup")

		topoStr := renderTopo(tidbNum, tikvNum, tiflashNum)
		deployTopo := fmt.Sprintf(tiupTopoTpl, arch, enableTls, topoStr)
		tc = tiup.NewTiUPCluster(controlNode, controlNodeName+"-peer", deployTopo, clusterName)
	})
	AfterEach(func() {
		err := UploadClinic(tc, "--include", "monitor,log", "-f", fmt.Sprintf("-%s", suiteTestCtx.Param("duration", "3h")))
		if err != nil {
			log.Error("upload clinic error", zap.Error(err))
		}
	})
	Context("run workload in cluster deployed with tiup", func() {
		It("should serve stably #001#", func() {
			tryDeployAndStartCluster(tc, version)
			wl = ch.NewWorkload(suiteTestCtx, tc, version, matcher.Try(strconv.ParseBool(enableTls)).(bool))
			wl.Prepare(suiteTestCtx)
			ctx0, _ := context.WithTimeout(suiteTestCtx, duration)
			g, ctx := errgroup.WithContext(ctx0)
			g.Go(util.WithGinkgoRecover(func() {
				matcher.Try(wl.RunTP(ctx, pkg.TPCommonErrHandler))
			}))
			g.Go(util.WithGinkgoRecover(func() {
				matcher.Try(wl.RunAP(ctx, pkg.APCommonErrHandler))
			}))
			matcher.Try(g.Wait())
		})
		It("should server stably when scale tiflash #002#", func() {
			tryDeployAndStartCluster(tc, version)
			wl = ch.NewWorkload(suiteTestCtx, tc, version, matcher.Try(strconv.ParseBool(enableTls)).(bool))
			wl.Prepare(suiteTestCtx)
			ctx0, _ := context.WithTimeout(suiteTestCtx, duration)
			g, ctx := errgroup.WithContext(ctx0)
			g.Go(util.WithGinkgoRecover(func() {
				matcher.Try(wl.RunTP(ctx, pkg.TPCommonErrHandler))
			}))
			g.Go(util.WithGinkgoRecover(func() {
				matcher.Try(wl.RunAP(ctx, pkg.APCommonErrHandler))
			}))
			g.Go(util.WithGinkgoRecover(func() {
				log.Info("begin scale out", zap.Int("tiflash nodes before scale", tiflashNum), zap.Int("tiflash scale out nodes", 2))
				matcher.Try(tc.ScaleOut(fmt.Sprintf(tiflashScaleOutTopo, tiflashNum+1, tiflashNum+2)))
				log.Info("end scale out", zap.Int("tiflash nodes before scale", tiflashNum), zap.Int("tiflash scale out nodes", 2))
				time.Sleep(90 * time.Minute)

				log.Info("begin scale in", zap.Int("tiflash nodes before scale", tiflashNum+2), zap.Strings("scale nodes", []string{
					fmt.Sprintf("tiflash-%d-peer:9000", tiflashNum+1), fmt.Sprintf("tiflash-%d-peer:9000", tiflashNum+2),
				}))
				matcher.Try(tc.ScaleIn(fmt.Sprintf("tiflash-%d-peer:9000", tiflashNum+1)))
				matcher.Try(tc.ScaleIn(fmt.Sprintf("tiflash-%d-peer:9000", tiflashNum+2)))
				log.Info("begin scale in", zap.Int("tiflash nodes before scale", tiflashNum+2), zap.Strings("scale nodes", []string{
					fmt.Sprintf("tiflash-%d-peer:9000", tiflashNum+1), fmt.Sprintf("tiflash-%d-peer:9000", tiflashNum+2),
				}))
				time.Sleep(60 * time.Minute)

				log.Info("destroy and remove instances that is in tombstone state")
				pruneCmd := workloadnode.CommandContext(suiteTestCtx, fmt.Sprintf("tiup cluster prune %s -y", clusterName))
				matcher.Try(pruneCmd.Run(controlNode))

			}))
			matcher.Try(g.Wait())
		})
	})
})

func tryDeployAndStartCluster(tc *tiup.TiUPCluster, version string) {
	info, err := tc.Display()
	if err == nil && info.ClusterMeta.ClusterName == clusterName {
		log.Info("cluster already deploy and start")
		return
	}
	log.Info("cluster not deploy and start and try to deploy and start", zap.Error(err))
	matcher.Try(tc.Deploy(version, "--format", "json"))
	log.Info("Cluster deployed")
	matcher.Try(tc.Start("--format", "json"))
	log.Info("Cluster started")
}