package scheduler

import (
	"math/rand"
	"net/http"
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"

	"github.com/pingcap/endless/pkg/client/pd"
	"github.com/pingcap/test-infra/sdk/pkg/util/pdutil"
)

func schedulerRoutine(pdcli *pd.Client, scheduler string, eg *errgroup.Group, closeCh <-chan struct{}) {
	eg.Go(func() error {
		log.Warn("INFO: SchedulerRoutine start", zap.String("scheduler", scheduler))
		defer func() {
			if err := pdcli.DeleteScheduler(scheduler); err != nil {
				log.Warn("clean scheduler error", zap.String("scheduler", scheduler), zap.Error(err))
			}
		}()

		if _, err := pdcli.CreateScheduler(scheduler); err != nil {
			log.Warn("scheduler create error", zap.String("scheduler", scheduler), zap.Error(err))
		}

		<-closeCh
		log.Warn("INFO: SchedulerRoutine end", zap.String("scheduler", scheduler))
		return nil
	})
}

func ShuffleLeaderRoutine(pdcli *pd.Client, eg *errgroup.Group, closeCh <-chan struct{}) {
	schedulerRoutine(pdcli, "shuffle-leader-scheduler", eg, closeCh)
}

func ShuffleRegionRoutine(pdcli *pd.Client, eg *errgroup.Group, closeCh <-chan struct{}) {
	schedulerRoutine(pdcli, "shuffle-region-scheduler", eg, closeCh)
}

func ClearTestingSchedulers(pdcli *pd.Client) {
	schedulers := []string{
		"shuffle-leader-scheduler",
		"shuffle-region-scheduler",
	}
	// TODO: query existed scheduler and delete
	for _, s := range schedulers {
		err := pdcli.DeleteScheduler(s)
		log.Info("clean scheduler", zap.String("scheduler", s), zap.Error(err))
	}
}

const (
	regionNumberLowerBound    = 20   // TODO: to be self-adaptive
	regionNumberUpperBound    = 2000 // TODO: to be self-adaptive
	regionChangePeriodSeconds = 10
	regionChangeFraction      = 0.3
)

func getRandomRegions(regions []*pdutil.RegionInfo, fraction float32) []*pdutil.RegionInfo {
	rand.Shuffle(len(regions), func(i, j int) {
		regions[i], regions[j] = regions[j], regions[i]
	})
	return regions[:int(float32(len(regions))*fraction)]
}

func newPDClient(pdAddr string) *pdutil.Client {
	return pdutil.NewPDClient(&http.Client{
		Timeout: 10 * time.Second,
	}, pdAddr)
}

type RegionChangeTaskFn func(pdcli *pdutil.Client, regions []*pdutil.RegionInfo) (expected_regions int, err error)

func splitRegion(pdcli *pdutil.Client, regions []*pdutil.RegionInfo) (expected_regions int, err error) {
	expected_regions = len(regions)
	if len(regions) < regionNumberUpperBound {
		success := 0
		for _, region := range getRandomRegions(regions, regionChangeFraction) {
			input := map[string]interface{}{
				"name":      "split-region",
				"region_id": region.ID,
				"policy":    "approximate", // alternative: "scan"
			}
			if err = pdcli.Operators(input); err == nil {
				success += 1
				log.Debug("split-region", zap.Any("input", input))
			} else {
				// Just ignore the error
				// "split-region" error happens when there is another operator running on this region
				log.Debug("split-region error", zap.Any("input", input), zap.Error(err))
			}
		}
		expected_regions += success
		log.Info("INFO: split-region", zap.Int("number of regions (expected)", expected_regions), zap.Int("success", success))
	}
	return expected_regions, nil
}

func mergeRegion(pdcli *pdutil.Client, regions []*pdutil.RegionInfo) (expected_regions int, err error) {
	expected_regions = len(regions)
	if len(regions) > regionNumberLowerBound {
		success := 0
		for _, region := range getRandomRegions(regions, regionChangeFraction) {
			siblings, err := pdcli.GetSiblingRegions(region.ID)
			if err != nil {
				return expected_regions, errors.Trace(err)
			}
			if len(siblings) > 0 {
				siblings = getRandomRegions(siblings, 1.0)
				input := map[string]interface{}{
					"name":             "merge-region",
					"source_region_id": region.ID,
					"target_region_id": siblings[0].ID,
				}
				if err = pdcli.Operators(input); err == nil {
					success += 1
					log.Debug("merge-region", zap.Any("input", input))
				} else {
					// Just ignore the error
					// "merge-region" error happens when there is another operator running on this region
					log.Debug("merge-region error", zap.Any("input", input), zap.Error(err))
				}
			}
		}
		expected_regions -= success
		log.Info("INFO: merge-region", zap.Int("number of regions (expected)", expected_regions), zap.Int("success", success))
	}
	return expected_regions, nil
}

func runRegionChangeTask(pdAddr string, eg *errgroup.Group, closeCh <-chan struct{}, taskName string, task RegionChangeTaskFn) {
	eg.Go(func() error {
		defer GinkgoRecover()
		log.Warn("INFO: runRegionChangeTask start", zap.String("task", taskName))
		for {
			select {
			case <-closeCh:
				log.Warn("INFO: runRegionChangeTask end", zap.String("task", taskName))
				return nil
			default:
			}

			randomSleepDuration := rand.Intn(regionChangePeriodSeconds)
			time.Sleep(time.Duration(randomSleepDuration) * time.Second)

			pdcli := newPDClient(pdAddr)

			regions, err := pdcli.ListRegions()
			if err != nil {
				log.Warn("runRegionChangeTask.ListRegions error", zap.String("task", taskName), zap.Error(err))
			} else {
				log.Info("INFO: runRegionChangeTask", zap.Int("regions", len(regions)), zap.String("task", taskName))
				log.Debug("runRegionChangeTask", zap.Any("regions", regions), zap.String("task", taskName))

				_, err = task(pdcli, regions)
				if err != nil {
					log.Warn("runRegionChangeTask task error", zap.String("task", taskName), zap.Error(err))
				}
			}

			time.Sleep(time.Duration(regionChangePeriodSeconds-randomSleepDuration) * time.Second)
		}
	})
}

type MergeSchedulerState int

const (
	MergeSchedulerOff MergeSchedulerState = iota
	MergeSchedulerOn

	MergeSchedulerName = "random-merge-scheduler"
)

// Deprecated. Use `runRegionChangeTask(mergeRegion)`.
func RegionRandomMergeRoutine(pdcli *pdutil.Client, eg *errgroup.Group, closeCh <-chan struct{}) {
	cleanUp := func() {
		// TODO: check scheduler existence by `pdcli.GetScheduler()`
		if err := pdcli.RemoveScheduler(MergeSchedulerName); err != nil {
			log.Warn("scheduler remove error", zap.String("scheduler", MergeSchedulerName), zap.Error(err))
		}
	}

	eg.Go(func() error {
		log.Warn("INFO: RegionRandomMergeRoutine start")
		defer GinkgoRecover()
		defer func() {
			cleanUp()
		}()
		cleanUp()
		state := MergeSchedulerOff

		for {
			regions, err := pdcli.ListRegions()
			Expect(err).Should(Succeed())
			log.Debug("RegionRandomMergeRoutine", zap.Any("regions", regions))

			if len(regions) > regionNumberLowerBound && state == MergeSchedulerOff {
				if err = pdcli.AddScheduler(MergeSchedulerName); err == nil {
					state = MergeSchedulerOn
				} else {
					log.Warn("scheduler create error", zap.String("scheduler", MergeSchedulerName), zap.Error(err))
				}
			} else if len(regions) <= regionNumberLowerBound && state == MergeSchedulerOn {
				if err = pdcli.RemoveScheduler(MergeSchedulerName); err == nil {
					state = MergeSchedulerOff
				} else {
					log.Warn("scheduler remove error", zap.String("scheduler", MergeSchedulerName), zap.Error(err))
				}
			}

			select {
			case <-closeCh:
				log.Warn("INFO: RegionRandomMergeRoutine end")
				return nil
			default:
			}

			time.Sleep(regionChangePeriodSeconds)
		}
	})
}

func RegionChangeRoutine(pdAddr string, eg *errgroup.Group, closeCh <-chan struct{}) {
	runRegionChangeTask(pdAddr, eg, closeCh, "split-region", splitRegion)
	runRegionChangeTask(pdAddr, eg, closeCh, "merge-region", mergeRegion)
}