package tikv_cdc

import (
	"context"
	"fmt"
	"math/rand"
	"net"
	"strings"
	"time"

	"github.com/pingcap/endless/testcase/rawkv/pkg/chaos"
	"github.com/pingcap/endless/testcase/rawkv/pkg/scheduler"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/pkg/util/pdutil"
	"go.uber.org/zap"

	pdclient "github.com/pingcap/endless/pkg/client/pd"
	"golang.org/x/sync/errgroup"
)

const (
	// newtwork latency
	TiKVCDCNetworkLatency = 120 * time.Millisecond
	TiKVNetworkLatency    = 10 * time.Millisecond
	// network latency duration
	NetworkLatencyDuration = 1 * time.Minute
	// network partition duration
	NetworkPartitionDuration = 30 * time.Second
	// network interruption duration
	NetworkInterruptDuration = 30 * time.Second
)

type ChaosFunc func(*ChaosRunner)

type ChaosEventType int

const (
	KillOneTiKV ChaosEventType = iota
	KillOnePD
	KillPDLeader
	KillOneTiKVCDC
	KillTiKVCDCOwner
	KillTiKVCDCProcessor
	// network latency
	DelayTiKV
	DelayTiKVCDC
	// network partition
	PartitionTiKV
	PartitionPD
	PartitionTiKVCDC
	// network interruption
	InterruptTiKV
	InterruptPD
	InterruptTiKVCDC
	// end
	ChaosEventCount
)

// region chaos include:
//    - Leader transfer
//    - Region split
//    - Region merge
// RegionChaos
type RegionScheduler struct {
	pdCli *pdclient.Client

	shuffleLeader bool
	shuffleRegion bool
	regionChange  bool

	eg      *errgroup.Group
	closeCh chan struct{}
}

func (r *RegionScheduler) StartSchedulerRoutine() {
	r.closeCh = make(chan struct{})

	if r.shuffleLeader {
		scheduler.ShuffleLeaderRoutine(r.pdCli, r.eg, r.closeCh)
	}
	if r.shuffleRegion {
		scheduler.ShuffleRegionRoutine(r.pdCli, r.eg, r.closeCh)
	}
	if r.regionChange {
		pdAddr := r.pdCli.GetPDAddr()
		scheduler.RegionChangeRoutine(pdAddr, r.eg, r.closeCh)
	}
}

func (r *RegionScheduler) StopSchedulerRoutine() {
	close(r.closeCh)
	r.eg.Wait()
	scheduler.ClearTestingSchedulers(r.pdCli)
}

func NewRegionScheduler(pdCli *pdclient.Client, shuffleLeader, shuffleRegion,
	regionChange bool) *RegionScheduler {
	r := &RegionScheduler{
		pdCli:         pdCli,
		shuffleLeader: shuffleLeader,
		shuffleRegion: shuffleRegion,
		regionChange:  regionChange,
		eg:            new(errgroup.Group),
	}
	return r
}

type ChaosRunner struct {
	clusterChaos chaos.Chaos
	regionChaos  *RegionScheduler

	cdcCli *TiKVCDCClient
	pdCli  *pdutil.Client

	chaosInterval       time.Duration
	regionChaosDuration time.Duration

	enableClusterChaos bool
	enableRegionChaos  bool
	clusterEventRandom bool
	nextClusterEvent   int

	ctx    context.Context
	cancel context.CancelFunc
	eg     *errgroup.Group
}

func NewChaosRunner(ctx context.Context, hostsFile, sshKeyFile, sshUser string,
	cdcCli *TiKVCDCClient, pdCli *pdclient.Client, chaosInterval time.Duration,
	regionChaosDuration time.Duration) (*ChaosRunner, error) {

	chaos, err := chaos.NewAnsibleChaos(hostsFile, sshKeyFile, sshUser, true)
	if err != nil {
		return nil, errors.Trace(err)
	}

	regionScheduler := NewRegionScheduler(pdCli, true, true, true)

	pdAddr := pdCli.GetPDAddr()
	nctx, cancel := context.WithCancel(ctx)
	cr := &ChaosRunner{
		clusterChaos: chaos,
		regionChaos:  regionScheduler,

		cdcCli: cdcCli,
		pdCli:  NewPDClient(pdAddr),

		chaosInterval:       chaosInterval,
		regionChaosDuration: regionChaosDuration,

		enableClusterChaos: true,
		enableRegionChaos:  true,

		clusterEventRandom: false,
		nextClusterEvent:   0,

		ctx:    nctx,
		cancel: cancel,
		eg:     new(errgroup.Group),
	}
	return cr, nil
}

func (c *ChaosRunner) StartChaos() {
	if c.enableClusterChaos {
		c.RunChaosRoutine(ClusterChaos)
	}
	if c.enableRegionChaos {
		c.RunChaosRoutine(RegionChaos)
	}
}

func (c *ChaosRunner) StopChaos() {
	c.cancel()
	c.eg.Wait()
}

func (c *ChaosRunner) SetClusterChaosRandom() {
	c.clusterEventRandom = true
}

func (c *ChaosRunner) RunChaosRoutine(fn ChaosFunc) {
	rand.Seed(time.Now().UnixNano())
	c.eg.Go(func() error {
		for {
			select {
			case <-c.ctx.Done():
				return nil
			case <-time.After(c.chaosInterval):
				fn(c)
			}
			select {
			case <-c.ctx.Done():
				return nil
			case <-time.After(time.Duration(c.regionChaosDuration.Seconds() * rand.Float64())):
			}
		}
	})
}

func RegionChaos(c *ChaosRunner) {
	c.regionChaos.StartSchedulerRoutine()
	defer c.regionChaos.StopSchedulerRoutine()
	log.Info("Start Region Chaos")

	select {
	case <-c.ctx.Done():
	case <-time.After(c.regionChaosDuration):
	}
}

func ClusterChaos(c *ChaosRunner) {
	var (
		err   error
		event int
	)
	if c.clusterEventRandom {
		event = rand.Int()
	} else {
		event = c.nextClusterEvent
		c.nextClusterEvent += 1
	}
	switch ChaosEventType(event) % ChaosEventCount {
	case KillOneTiKV:
		err = c.KillOneTiKV()
	case KillOnePD:
		err = c.KillOnePD()
	case KillPDLeader:
		err = c.KillPDLeader()
	case KillOneTiKVCDC:
		err = c.KillOneTiKVCDC()
	case KillTiKVCDCOwner:
		err = c.KillTiKVCDCOwner()
	case KillTiKVCDCProcessor:
		err = c.KillTiKVCDCProcessor()
	case DelayTiKV:
		err = c.DelayTiKV()
	case DelayTiKVCDC:
		err = c.DelayTiKVCDC()
	case PartitionTiKV:
		err = c.PartitionTiKV()
	case PartitionPD:
		err = c.PartitionPD()
	case PartitionTiKVCDC:
		err = c.PartitionTiKVCDC()
	case InterruptTiKV:
		err = c.InterruptTiKV()
	case InterruptPD:
		err = c.InterruptPD()
	case InterruptTiKVCDC:
		err = c.InterruptTiKVCDC()
	}
	if err != nil {
		log.Warn("Faile to simulate chaos event", zap.Error(err), zap.Any("event", event))
	}
}

func (c *ChaosRunner) KillOneTiKV() error {
	log.Info("Try to kill one TiKV")
	return c.clusterChaos.KillOneTiKV()
}

func (c *ChaosRunner) KillOnePD() error {
	log.Info("Try to kill one PD")
	return c.clusterChaos.KillOnePD()
}

func (c *ChaosRunner) KillOneTiKVCDC() error {
	log.Info("Try to kill one TiKVCDC")
	return c.clusterChaos.ChaosAction("tikv_cdc", chaos.AnyOne, chaos.KillServer, nil, 0, 0)
}

func (c *ChaosRunner) KillPDLeader() error {
	log.Info("Try to kill PD leader")
	m, err := c.pdCli.GetMembers()
	if err != nil {
		return err
	}
	leaderName := strings.Split(m.Leader.Name, "-")
	if len(leaderName) != 3 {
		return fmt.Errorf("Failed to get PD address from PD leader name, PD leader name: %s", m.Leader.Name)
	}
	leaderHost := leaderName[1]
	return c.clusterChaos.ChaosAction("pd", chaos.GivenOne, chaos.KillServer, []string{leaderHost}, 0, 0)
}

func (c *ChaosRunner) KillTiKVCDCOwner() error {
	log.Info("Try to kill TiKVCDC Owner")
	captures, err := c.cdcCli.TiKVCDCCaptureList(c.ctx)
	if err != nil {
		return err
	}
	ownerAddr := ""
	for _, capture := range captures {
		if capture.IsOwner {
			ownerAddr = capture.Address
			break
		}
	}
	if ownerAddr == "" {
		return fmt.Errorf("Failed to get owner address")
	}
	ownerHost, _, err := net.SplitHostPort(ownerAddr)
	if err != nil {
		return err
	}
	return c.clusterChaos.ChaosAction("tikv_cdc", chaos.GivenOne, chaos.KillServer, []string{ownerHost}, 0, 0)
}

func (c *ChaosRunner) KillTiKVCDCProcessor() error {
	log.Info("Try to kill TiKVCDC Processor")
	processors, err := c.cdcCli.TiKVCDCProcessorList(c.ctx)
	if err != nil {
		return err
	}
	if len(processors) != 1 {
		return fmt.Errorf("No processor working")
	}
	captureID := processors[0].CaptureId

	captures, err := c.cdcCli.TiKVCDCCaptureList(c.ctx)
	if err != nil {
		return err
	}
	processorAddr := ""
	for _, capture := range captures {
		if capture.ID == captureID {
			processorAddr = capture.Address
			break
		}
	}
	if processorAddr == "" {
		return fmt.Errorf("Failed to get processor address")
	}
	processorHost, _, err := net.SplitHostPort(processorAddr)
	if err != nil {
		return err
	}
	return c.clusterChaos.ChaosAction("tikv_cdc", chaos.GivenOne, chaos.KillServer, []string{processorHost}, 0, 0)
}

func (c *ChaosRunner) DelayTiKVCDC() error {
	log.Info("Try to delay TiKVCDC cluster newtork")
	return c.clusterChaos.ChaosAction("tikv_cdc", chaos.AllMembers, chaos.DelayNetwork, nil, TiKVCDCNetworkLatency, NetworkLatencyDuration)
}

func (c *ChaosRunner) DelayTiKV() error {
	log.Info("Try to delay TiKV cluster newtork")
	return c.clusterChaos.ChaosAction("tikv", chaos.AllMembers, chaos.DelayNetwork, nil, TiKVNetworkLatency, NetworkLatencyDuration)
}

func (c *ChaosRunner) PartitionTiKV() error {
	log.Info("Try to partition TiKV cluster newtork")
	return c.clusterChaos.ChaosAction("tikv", chaos.AnyOne, chaos.InterruptNetwork, nil, 0, NetworkPartitionDuration)
}

func (c *ChaosRunner) PartitionPD() error {
	log.Info("Try to partition PD cluster newtork")
	return c.clusterChaos.ChaosAction("pd", chaos.AnyOne, chaos.InterruptNetwork, nil, 0, NetworkPartitionDuration)
}

func (c *ChaosRunner) PartitionTiKVCDC() error {
	log.Info("Try to partition TiKVCDC cluster newtork")
	return c.clusterChaos.ChaosAction("tikv_cdc", chaos.AnyOne, chaos.InterruptNetwork, nil, 0, NetworkPartitionDuration)
}

func (c *ChaosRunner) InterruptTiKV() error {
	log.Info("Try to interrupt TiKV cluster network")
	return c.clusterChaos.ChaosAction("tikv", chaos.AllMembers, chaos.InterruptNetwork, nil, 0, NetworkInterruptDuration)
}

func (c *ChaosRunner) InterruptPD() error {
	log.Info("Try to interrupt PD cluster network")
	return c.clusterChaos.ChaosAction("pd", chaos.AllMembers, chaos.InterruptNetwork, nil, 0, NetworkInterruptDuration)
}

func (c *ChaosRunner) InterruptTiKVCDC() error {
	log.Info("Try to interrupt TiKVCDC cluster network")
	return c.clusterChaos.ChaosAction("tikv_cdc", chaos.AllMembers, chaos.InterruptNetwork, nil, 0, NetworkInterruptDuration)
}