package tikv_cdc

import (
	"context"
	"flag"
	"time"

	pdclient "github.com/pingcap/endless/pkg/client/pd"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	_ "github.com/pingcap/endless/pkg/client/tidb"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
)

// TODO:
// 1. Modify network delay
// 2. Modify network interruption
// 3. Add chaos execution success confirmation mechanism
// 4. Replice tikv client retry with pkg/client

const (
	Uint64Max = ^uint64(0)
)

// Example of running with parameters:
// `ginkgo -focus rawkv_cdc_basic_v1 -- TiKVCDC -threads=50 -keys=10000`
var _ = ParameterizedDescribe("TiKVCDC", func(flagSet *flag.FlagSet) {
	ctx := suiteTestCtx

	var (
		flagThreads             int64
		flagKeyCnts             int64
		flagDuration            time.Duration
		flagRunOnce             bool
		flagOpDuration          time.Duration
		flagCdcPath             string
		flagChangefeedId        string
		flagSortEngine          string
		flagWaitSinkTimeout     time.Duration
		flagCaptureCnts         int
		flagAnsibleHosts        string
		flagAnsibleKey          string
		flagAnsibleUser         string
		flagChaosInterval       time.Duration
		flagRegionChaosDuration time.Duration

		srcTc     resource.TiDBCluster
		dstTc     resource.TiDBCluster
		cdc       resource.WorkloadNode
		srcPDAddr string
		dstPDAddr string

		cdcClient   *TiKVCDCClient
		srcPDClient *pdclient.Client
	)

	flagSet.Int64Var(&flagThreads, "threads", 500, "Number of threads")
	flagSet.Int64Var(&flagKeyCnts, "keys", 10000000, "Number of preload keys")
	flagSet.DurationVar(&flagDuration, "duration", 24*time.Hour, "Duration (e.g. 24h) to run")
	flagSet.DurationVar(&flagOpDuration, "op-duration", 3*time.Minute, "Duration (e.g. 3m) to run workload")
	flagSet.BoolVar(&flagRunOnce, "run-once", true, "Test case to run once")
	flagSet.StringVar(&flagCdcPath, "cdc-path", "tikv-cdc", "Path to TiKVCDC")
	flagSet.StringVar(&flagChangefeedId, "changefeed-id", "feed01", "The changefeed ID")
	flagSet.StringVar(&flagSortEngine, "sort-engine", "unified", "The changefeed sort engine")
	flagSet.DurationVar(&flagWaitSinkTimeout, "wait-sink-timeout", 10*time.Minute, "Duration (e.g. 10m) to wait tikv-cdc sink")
	flagSet.IntVar(&flagCaptureCnts, "captures", 3, "Number of captures")
	flagSet.StringVar(&flagAnsibleHosts, "hosts", "hosts", "The hosts file")
	flagSet.StringVar(&flagAnsibleKey, "ssh-key", "", "The ssh key file")
	flagSet.StringVar(&flagAnsibleUser, "ssh-user", "root", "The ssh user")
	flagSet.DurationVar(&flagChaosInterval, "chaos-interval", 5*time.Minute, "Min interval (e.g. 5m) to happen chaos")
	flagSet.DurationVar(&flagRegionChaosDuration, "region-chaos-duration", 3*time.Minute, "Duratirn(e.g. 3m) to run region chaos")

	BeforeEach(func() {
		srcTc = ctx.Resource("srctc").(resource.TiDBCluster)
		dstTc = ctx.Resource("dsttc").(resource.TiDBCluster)
		cdc = ctx.Resource("cdc").(resource.WorkloadNode)

		srcPDClient = Try(pdclient.NewPDClient(srcTc)).(*pdclient.Client)
		srcPDAddr = srcPDClient.GetPDAddr()

		dstPDClient := Try(pdclient.NewPDClient(dstTc)).(*pdclient.Client)
		dstPDAddr = dstPDClient.GetPDAddr()

		cdcClient = NewTiKVCDCClient(cdc, flagCdcPath, srcPDAddr)

		log.Info("BeforeEach", zap.Int64("keys", flagKeyCnts),
			zap.Duration("duration", flagDuration), zap.String("srcPDAddr", srcPDAddr),
			zap.String("dstPDAddr", dstPDAddr), zap.String("cdcPath", flagCdcPath),
			zap.String("changefeedId", flagChangefeedId))
	})

	runLoop := func(loop func(ctx context.Context, cfRunner *ChangefeedRunner,
		wlRunner *WorkloadRunner, workload Workload), startChaos,
		createChangeFeedBefore, setTimeSleep, safePointCheck bool) {

		nctx, cancel := context.WithCancel(ctx)
		defer cancel()
		cfRunner, err := NewChangefeedRunner(nctx, cdcClient, flagChangefeedId, dstPDAddr, flagSortEngine)
		Expect(err).Should(Succeed())
		defer cfRunner.Close()

		wlRunner, err := NewWorkloadRunner(nctx, srcPDAddr, dstPDAddr, flagKeyCnts, flagThreads, flagWaitSinkTimeout)
		Expect(err).Should(Succeed())
		defer wlRunner.Close()

		cr, err := NewChaosRunner(nctx, flagAnsibleHosts, flagAnsibleKey, flagAnsibleUser, cdcClient, srcPDClient, flagChaosInterval, flagRegionChaosDuration)
		Expect(err).Should(Succeed())
		if startChaos {
			cr.StartChaos()
			defer cr.StopChaos()
		}

		if createChangeFeedBefore {
			if safePointCheck {
				srcPDClient := NewPDClient(srcPDAddr)
				startTs, err := cdcClient.TiKVCDCTsoQuery(nctx)
				Expect(err).Should(Succeed())
				// sleep 1s to wait causal ts flush.
				time.Sleep(1 * time.Second)

				cfRunner.Options.StartTs = startTs
				cfRunner.StartSafePointCheck(startTs, srcPDClient)
			}
			cfRunner.ChangefeedOpHelper(nctx, Create)
		}

		if setTimeSleep {
			keyCntPerBatch := flagKeyCnts / flagThreads
			timeSleepMs := flagOpDuration.Milliseconds() / keyCntPerBatch
			wlRunner.SetTimeSleep(timeSleepMs)
		}

		timer := time.NewTimer(flagDuration)
		for {
			select {
			case <-nctx.Done():
				break
			case <-timer.C:
				break
			default:
				for _, workload := range Workloads {
					loop(nctx, cfRunner, wlRunner, workload)
				}
				wlRunner.SetAtomicForCAS(true)
				for _, workload := range AtomicWorkloads {
					loop(nctx, cfRunner, wlRunner, workload)
				}
				wlRunner.SetAtomicForCAS(false)
			}
			if flagRunOnce && !startChaos {
				break
			}
		}
	}

	Context("ApiV2", func() {
		TestCase("TiKVCDC", "rawkv_cdc_test_1.1", func() {
			loop := func(ctx context.Context, cfRunner *ChangefeedRunner,
				wlRunner *WorkloadRunner, workload Workload) {
				err := wlRunner.RunWorkload(ctx, workload, DoAndVerfiy)
				Expect(err).Should(Succeed())
			}

			runLoop(loop, false, true, false, false)
		})

		TestCase("TiKVCDC", "rawkv_cdc_test_1.2", func() {
			loop := func(ctx context.Context, cfRunner *ChangefeedRunner,
				wlRunner *WorkloadRunner, workload Workload) {
				startTs, err := cdcClient.TiKVCDCTsoQuery(ctx)
				Expect(err).Should(Succeed())
				cfRunner.Options.StartTs = startTs
				// sleep 1s to wait causal ts flush.
				time.Sleep(1 * time.Second)

				err = wlRunner.RunWorkload(ctx, workload, DoWorkload)
				Expect(err).Should(Succeed())
				cfRunner.ChangefeedOpHelper(ctx, Create)

				err = wlRunner.RunWorkload(ctx, workload, Verify)
				Expect(err).Should(Succeed())
				cfRunner.ChangefeedOpHelper(ctx, Remove)
			}
			runLoop(loop, false, false, false, false)
		})

		TestCase("TiKVCDC", "rawkv_cdc_test_1.3", func() {
			// OpDuration must must last longer than 3m
			Expect(flagOpDuration.Minutes()).Should(BeNumerically(">=", 3.0))

			loop := func(ctx context.Context, cfRunner *ChangefeedRunner,
				wlRunner *WorkloadRunner, workload Workload) {
				go func() {
					time.Sleep(flagOpDuration / 3)
					cfRunner.ChangefeedOpHelper(ctx, Pause)
					log.Info("Pause Changefeed")

					time.Sleep(flagOpDuration / 3)
					cfRunner.ChangefeedOpHelper(ctx, Resume)
					log.Info("Resume Changefeed")
				}()

				err := wlRunner.RunWorkload(ctx, workload, DoAndVerfiy)
				Expect(err).Should(Succeed())
			}
			runLoop(loop, false, true, true, false)
		})

		TestCase("TiKVCDC", "rawkv_cdc_test_1.5", func() {
			// OpDuration must last longer than 1m
			Expect(flagOpDuration.Minutes()).Should(BeNumerically(">=", 1.0))

			loop := func(ctx context.Context, cfRunner *ChangefeedRunner,
				wlRunner *WorkloadRunner, workload Workload) {
				err := wlRunner.RunWorkload(ctx, workload, DoAndVerfiy)
				Expect(err).Should(Succeed())
			}
			runLoop(loop, false, true, true, true)
		})

		TestCase("TiKVCDC", "rawkv_cdc_test_1.5_failed1", func() {
			nctx, cancel := context.WithCancel(ctx)
			defer cancel()
			cfRunner, err := NewChangefeedRunner(nctx, cdcClient, flagChangefeedId, dstPDAddr, flagSortEngine)
			Expect(err).Should(Succeed())
			defer cfRunner.Close()

			srcPDClient := NewPDClient(srcPDAddr)
			startTs, err := GetServiceSafePoint(srcPDClient, GCSafePointID)
			Expect(err).Should(Succeed())
			if startTs == 0 {
				startTs, err = GetServiceSafePoint(srcPDClient, GCWorkerServiceSafePointID)
				Expect(err).Should(Succeed())
			}
			cfRunner.Options.StartTs = startTs - 1

			// Command `tikv-cdc cli changefeed create` maybe return err
			err = cdcClient.TiKVCDCCreateChangefeed(nctx, flagChangefeedId, cfRunner.Options)
			if err == nil {
				err = cdcClient.ChangefeedStateCheck(nctx, flagChangefeedId, "failed")
				Expect(err).Should(Succeed())
			}
		})

		TestCase("TiKVCDC", "rawkv_cdc_test_1.5_failed2", func() {
			// This test case relies on setting tikv-cdc server gc-ttl to 300 (5 * 60s).
			nctx, cancel := context.WithCancel(ctx)
			defer cancel()
			cfRunner, err := NewChangefeedRunner(nctx, cdcClient, flagChangefeedId, dstPDAddr, flagSortEngine)
			Expect(err).Should(Succeed())
			defer cfRunner.Close()

			srcPDClient := NewPDClient(srcPDAddr)
			cfRunner.ChangefeedOpHelper(nctx, Create)
			cfRunner.ChangefeedOpHelper(nctx, Pause)
			time.Sleep(WaitCdcExecuteTime)

			log.Info("Wait until the cdc safepoint blocks the gc safe point...")
			err = Retry(24, 1*time.Minute, func() (bool, error) {
				gcSafePoint, err := GetServiceSafePoint(srcPDClient, GCSafePointID)
				Expect(err).Should(Succeed())
				cdcSafePoint, err := GetServiceSafePoint(srcPDClient, TiKVCDCServiceSafePointID)
				Expect(err).Should(Succeed())
				log.Info("Now the service safe point", zap.Uint64("gcSafePoint", gcSafePoint),
					zap.Uint64("cdcSafePoint", cdcSafePoint))
				return gcSafePoint == cdcSafePoint || cdcSafePoint == Uint64Max-1, nil
			})
			Expect(err).Should(Succeed())

			// Command `tikv-cdc cli changefeed resume` maybe return err
			err = cdcClient.TiKVCDCResumeChangefeed(nctx, flagChangefeedId)
			if err == nil {
				err = cdcClient.ChangefeedStateCheck(nctx, flagChangefeedId, "failed")
				Expect(err).Should(Succeed())
			}
		})

		TestCase("TiKVCDC", "rawkv_cdc_test_2", func() {
			nctx, cancel := context.WithCancel(ctx)
			defer cancel()
			cfRunner, err := NewChangefeedRunner(nctx, cdcClient, flagChangefeedId, dstPDAddr, flagSortEngine)
			Expect(err).Should(Succeed())
			defer cfRunner.Close()

			captureInfos, err := cdcClient.TiKVCDCCaptureList(nctx)
			Expect(err).Should(Succeed())
			Expect(len(captureInfos)).Should(Equal(flagCaptureCnts))

			processorInfos, err := cdcClient.TiKVCDCProcessorList(nctx)
			Expect(err).Should(Succeed())
			Expect(len(processorInfos)).Should(Equal(0))

			cfRunner.ChangefeedOpHelper(nctx, Create)
			time.Sleep(WaitCdcExecuteTime)

			changefeedInfos, err := cdcClient.TiKVCDCChangefeedList(nctx)
			Expect(err).Should(Succeed())
			Expect(len(changefeedInfos)).Should(Equal(1))

			processorInfos, err = cdcClient.TiKVCDCProcessorList(nctx)
			Expect(err).Should(Succeed())
			Expect(len(processorInfos)).Should(Equal(1))

			changefeedId, captureId := processorInfos[0].ChangefeedId, processorInfos[0].CaptureId
			_, err = cdcClient.TiKVCDCProcessorQuery(nctx, changefeedId, captureId)
			Expect(err).Should(Succeed())

			cfRunner.ChangefeedOpHelper(nctx, Pause)
			cfRunner.ChangefeedOpHelper(nctx, Resume)
			cfRunner.ChangefeedOpHelper(nctx, Remove)
			time.Sleep(WaitCdcExecuteTime)

			changefeedInfos, err = cdcClient.TiKVCDCChangefeedList(nctx)
			Expect(err).Should(Succeed())
			Expect(len(changefeedInfos)).Should(Equal(0))
		})

		TestCase("TiKVCDC", "rawkv_cdc_stability_test", func() {
			loop := func(ctx context.Context, cfRunner *ChangefeedRunner,
				wlRunner *WorkloadRunner, workload Workload) {
				err := wlRunner.RunWorkload(ctx, workload, DoAndVerfiy)
				Expect(err).Should(Succeed())
			}

			runLoop(loop, true, true, false, false)
		})
	})
})