package steps

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/caselib/pkg/utils"
	apiresource "github.com/pingcap/test-infra/common/model/resource"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
	sdkresource "github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
)

func NewEnableTiDBGeneralLog(handle *apiresource.ResourceHandle) Step {
	return &enableTiDBGeneralLog{
		handle: handle,
	}
}

type enableTiDBGeneralLog struct {
	handle *apiresource.ResourceHandle
}

func (s *enableTiDBGeneralLog) String() string {
	return fmt.Sprintf("enableTiDBGeneralLog, target=%+v", s.handle)
}

func (s *enableTiDBGeneralLog) Execute(ctx context.Context) error {
	return NewExecuteSQL(s.handle, "", "set @@tidb_general_log=on").Execute(ctx)
}

func NewGCLifeTime(handle *apiresource.ResourceHandle, gcLifeTime string) Step {
	return &setGCLifeTime{
		handle:     handle,
		gcLifeTime: gcLifeTime,
	}
}

type setGCLifeTime struct {
	handle     *apiresource.ResourceHandle
	gcLifeTime string
}

func (s *setGCLifeTime) String() string {
	return fmt.Sprintf("setGCLifeTime, target=%+v, value=%s", s.handle, s.gcLifeTime)
}

func (s *setGCLifeTime) Execute(ctx context.Context) error {
	return NewExecuteSQL(s.handle, "", fmt.Sprintf("set GLOBAL tidb_gc_life_time ='%s'", s.gcLifeTime)).Execute(ctx)
}

func NewMaxConnection(handle *apiresource.ResourceHandle, maxConnection int) Step {
	return &setMaxConnection{
		handle:     handle,
		connection: maxConnection,
	}
}

type setMaxConnection struct {
	handle     *apiresource.ResourceHandle
	connection int
}

func (s *setMaxConnection) String() string {
	return fmt.Sprintf("setMaxConnection, target=%+v, value=%d", s.handle, s.connection)
}

func (s *setMaxConnection) Execute(ctx context.Context) error {
	return NewExecuteSQL(s.handle, "", fmt.Sprintf("set GLOBAL max_connections =%d", s.connection)).Execute(ctx)
}

type scaleTiDBCluster struct {
	target  resource.TiDBCompType
	replica uint32
	timeout time.Duration
}

func (s *scaleTiDBCluster) String() string {
	return fmt.Sprintf("scaleTiDBCluster, target=%+v, replica=%+v, timeout=%+v", s.target, s.replica, s.timeout)
}

func (s *scaleTiDBCluster) Execute(ctx context.Context) error {
	tc, err := GetInner(ctx).Board().TiDBCluster()
	if err != nil {
		return err
	}

	start := time.Now()
	if err := tc.Scale(resource.TiDBClusterScaleOptions{
		Target:   s.target,
		Replicas: s.replica,
		Timeout:  s.timeout,
	}); err != nil {
		log.Error("scale tidb cluster failed", zap.Error(err), zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("scale tidb cluster success", zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))
	return nil
}

func NewScaleTiDBCluster(tp resource.TiDBCompType, replica uint32, timeout time.Duration) Step {
	return &scaleTiDBCluster{
		target:  tp,
		replica: replica,
		timeout: timeout,
	}
}

type upgradeTiDBCluster struct {
	target  resource.TiDBCompType
	version *string
	timeout time.Duration
}

func (s *upgradeTiDBCluster) String() string {
	return fmt.Sprintf("upgradeTiDBCluster, target=%+v, version=%+v, timeout=%+v", s.target, *s.version, s.timeout)
}

func (s *upgradeTiDBCluster) Execute(ctx context.Context) error {
	tc, err := GetInner(ctx).Board().TiDBCluster()
	if err != nil {
		return err
	}

	start := time.Now()
	if err := tc.Upgrade(resource.TiDBClusterUpgradeOptions{
		ComponentUpgradeOption: &resource.TiDBClusterComponentUpgradeOption{
			Target:  s.target,
			Version: *s.version,
		},
		Timeout: s.timeout,
	}); err != nil {
		log.Error("upgrade tidb cluster failed", zap.Error(err), zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("upgrade tidb cluster success", zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))

	return nil
}

func NewUpgradeTiDBCluster(tp resource.TiDBCompType, version *string, timeout time.Duration) Step {
	return &upgradeTiDBCluster{
		target:  tp,
		version: version,
		timeout: timeout,
	}
}

type updateConfig struct {
	target  resource.TiDBCompType
	config  string
	timeout time.Duration
}

func (s *updateConfig) String() string {
	return fmt.Sprintf("updateConfig, target=%+v, config=%+v, timeout=%+v", s.target, s.config, s.timeout)
}

func (s *updateConfig) Execute(ctx context.Context) error {
	tc, err := GetInner(ctx).Board().TiDBCluster()
	if err != nil {
		return err
	}

	start := time.Now()
	log.Info("Start to update tc config!!!", zap.Any("option", s))
	if err := tc.UpdateConfig(resource.TiDBClusterUpdateConfigOptions{
		Target:  s.target,
		Config:  s.config,
		Timeout: s.timeout,
	}); err != nil {
		log.Error("update tc config failed", zap.Error(err), zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("update tc config success", zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))

	return nil
}

func NewUpdateConfig(tp resource.TiDBCompType, config string, timeout time.Duration) Step {
	return &updateConfig{
		target:  tp,
		config:  config,
		timeout: timeout,
	}
}

// initialize tidb-cluster for scale-out/scale-in
type scaleTiDBcomponent struct {
	tidb    *apiresource.ResourceHandle
	tp      sdkresource.TiDBCompType
	replica uint32
	timeout time.Duration
}

func NEWScaleTiDBcomponent(tidbcluster *apiresource.ResourceHandle, tp sdkresource.TiDBCompType, replica uint32, timeout time.Duration) Step {
	return &scaleTiDBcomponent{
		tidb:    tidbcluster,
		tp:      tp,
		replica: replica,
		timeout: timeout,
	}
}

func (s *scaleTiDBcomponent) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(s.tidb)
	if err != nil {
		return errors.Trace(err)
	}
	initTidb := resolved.(resource.TiDBCluster)

	mc, err := GetInner(ctx).Board().TiDBCluster()
	if mc == nil {
		if err := GetInner(ctx).Board().SetTiDBCluster(initTidb); err != nil {
			return errors.Trace(err)
		}
	}

	tc, err := GetInner(ctx).Board().TiDBCluster()
	if err != nil {
		return err
	}

	start := time.Now()
	if err := tc.Scale(resource.TiDBClusterScaleOptions{
		Target:   s.tp,
		Replicas: s.replica,
		Timeout:  s.timeout,
	}); err != nil {
		log.Error("scale tidb cluster failed", zap.Error(err), zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("scale tidb cluster success", zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))
	return nil
}

func (s *scaleTiDBcomponent) String() string {
	return fmt.Sprintf("scale TiDB component: TiDB=%+v", s.tp)
}

// initialize tidb-cluster for upgrade
type ugradeTiDBAllCcomponent struct {
	tidb    *apiresource.ResourceHandle
	version *string
	timeout time.Duration
}

func NewUpgradeTiDBAllComponent(tidbcluster *apiresource.ResourceHandle, version *string, timeout time.Duration) Step {
	return &ugradeTiDBAllCcomponent{
		tidb:    tidbcluster,
		version: version,
		timeout: timeout,
	}
}

func (s *ugradeTiDBAllCcomponent) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(s.tidb)
	if err != nil {
		return errors.Trace(err)
	}
	initTidb := resolved.(resource.TiDBCluster)

	mc, err := GetInner(ctx).Board().TiDBCluster()
	if mc == nil {
		if err := GetInner(ctx).Board().SetTiDBCluster(initTidb); err != nil {
			return errors.Trace(err)
		}
	}

	tc, err := GetInner(ctx).Board().TiDBCluster()
	if err != nil {
		return err
	}

	// Upgrade tidb
	start := time.Now()
	// check whether current version already be target version, and skip upgrade (otherwise upgrade might hang).
	status, err := tc.GetStatus(sdkresource.TiDB)
	if err != nil {
		return errors.Trace(err)
	} else {
		log.Debug(status.Image)
		currentVer := strings.Split(status.Image, ":")[1]
		log.Info(fmt.Sprintf("Current TiDB Cluster version: %s", currentVer))
		if currentVer == *s.version {
			log.Info("TiDB Cluster version already matches target version, skip upgrade TiDB.")
			return nil
		}
	}
	if err := tc.Upgrade(resource.TiDBClusterUpgradeOptions{
		Version: *s.version,
		Timeout: s.timeout,
	}); err != nil {
		log.Error("upgrade tidb cluster failed", zap.Error(err), zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("upgrade tidb cluster success", zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))

	return nil
}

func (s *ugradeTiDBAllCcomponent) String() string {
	versionStr := "<<???>>"
	if s.version != nil {
		versionStr = *s.version
	}
	return fmt.Sprintf("Upgrade TiDB All Component to version=%s", versionStr)
}

type updateTiDBClusterConfig struct {
	target   *apiresource.ResourceHandle
	nodeType sdkresource.TiDBCompType
	config   string
	timeout  time.Duration
}

func (u *updateTiDBClusterConfig) String() string {
	return fmt.Sprintf("kvRegionSplit")
}

func (u *updateTiDBClusterConfig) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(u.target)
	if err != nil {
		return errors.Trace(err)
	}
	initTidb := resolved.(resource.TiDBCluster)

	mc, err := GetInner(ctx).Board().TiDBCluster()
	if mc == nil {
		if err := GetInner(ctx).Board().SetTiDBCluster(initTidb); err != nil {
			return errors.Trace(err)
		}
	}

	tc, err := GetInner(ctx).Board().TiDBCluster()
	if err != nil {
		return err
	}

	start := time.Now()
	log.Info("Start to update TiDB Cluster config!!!")
	if err := tc.UpdateConfig(resource.TiDBClusterUpdateConfigOptions{
		Target:  u.nodeType,
		Config:  u.config,
		Timeout: u.timeout,
	}); err != nil {
		log.Error("Update TiDB Cluster config failed", zap.Error(err), zap.Any("option", u), zap.Duration("elapsed", time.Since(start)))
		if err.Error() == "no updated config" {
			log.Warn("Update TiDB cluster config failed due to no updated config. Ignore it.")
			return nil
		}
		return errors.Trace(err)
	}

	log.Info("Update TiDB Cluster config success", zap.Any("option", u), zap.Duration("elapsed", time.Since(start)))

	return nil
}

func NewUpdateTiDBClusterConfig(target *apiresource.ResourceHandle, nodeType sdkresource.TiDBCompType, config string, timeout time.Duration) Step {
	return &updateTiDBClusterConfig{
		target:   target,
		nodeType: nodeType,
		config:   config,
		timeout:  timeout,
	}
}

func (b *Builder) UpdateTiDBClusterConfig(target *apiresource.ResourceHandle, nodeType sdkresource.TiDBCompType, config string, timeout time.Duration) *Builder {
	b.steps = append(b.steps, NewUpdateTiDBClusterConfig(target, nodeType, config, timeout))
	return b
}

type dmUpgradeTiDB struct {
	tidb    *apiresource.ResourceHandle
	version *string
	timeout time.Duration
}

func (b *Builder) DmUpgradeTiDB(tidbcluster *apiresource.ResourceHandle, version *string, timeout time.Duration) *Builder {
	b.steps = append(b.steps, &dmUpgradeTiDB{
		tidb:    tidbcluster,
		version: version,
		timeout: timeout,
	})
	return b
}

func (s *dmUpgradeTiDB) Execute(ctx context.Context) error {
	initTidb := utils.Try(core.ResolveResourceHandle(s.tidb)).(resource.TiDBCluster)
	// Upgrade tidb
	start := time.Now()
	if err := initTidb.Upgrade(resource.TiDBClusterUpgradeOptions{
		ComponentUpgradeOption: &resource.TiDBClusterComponentUpgradeOption{
			Version: *s.version,
		},
		Timeout: s.timeout,
	}); err != nil {
		log.Error("upgrade tidb cluster failed", zap.Error(err), zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("upgrade tidb cluster success", zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))

	return nil
}

func (s *dmUpgradeTiDB) String() string {
	versionStr := "<<???>>"
	if s.version != nil {
		versionStr = *s.version
	}
	return fmt.Sprintf("Upgrade TiDB All Component to version=%s", versionStr)
}

type restartTiDBComponent struct {
	tidbRes  *apiresource.ResourceHandle
	nodeType sdkresource.TiDBCompType
	timeout  time.Duration
}

func (r restartTiDBComponent) String() string {
	return fmt.Sprintf("restart tidb component")
}

func (r restartTiDBComponent) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(r.tidbRes)
	if err != nil {
		return errors.Trace(err)
	}
	initTidb := resolved.(resource.TiDBCluster)

	mc, err := GetInner(ctx).Board().TiDBCluster()
	if mc == nil {
		if err := GetInner(ctx).Board().SetTiDBCluster(initTidb); err != nil {
			return errors.Trace(err)
		}
	}

	tc, err := GetInner(ctx).Board().TiDBCluster()
	if err != nil {
		return err
	}
	nodes, err := tc.GetNodes(r.nodeType)
	if err != nil {
		return err
	}
	for _, node := range nodes {
		if err := node.Restart(); err != nil {
			return err
		}
	}
	return nil
}

func (b *Builder) RestartTiDBComponent(tidbRes *apiresource.ResourceHandle, nodeType sdkresource.TiDBCompType, timeout time.Duration) *Builder {
	b.steps = append(b.steps, NewRestartTiDBComponent(tidbRes, nodeType, timeout))
	return b
}
func NewRestartTiDBComponent(tidbRes *apiresource.ResourceHandle, nodeType sdkresource.TiDBCompType, timeout time.Duration) Step {
	return &restartTiDBComponent{tidbRes: tidbRes, nodeType: nodeType, timeout: timeout}
}

func NewWatchCDCTiKVOOM(handle *apiresource.ResourceHandle) Step {
	return &watchCDCTiKVOOM{
		handle: handle,
	}
}

type watchCDCTiKVOOM struct {
	handle *apiresource.ResourceHandle
}

func (s *watchCDCTiKVOOM) String() string {
	return fmt.Sprintf("watchCDCTiKVOOM, cluster=%+v", s.handle)
}

func (s *watchCDCTiKVOOM) Execute(ctx context.Context) error {
	tc, err := GetInner(ctx).Board().TiDBCluster()
	if err != nil {
		return err
	}
	cancelCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	cdcEventChan, err := tc.WatchOOM(cancelCtx, sdkresource.TiDBClusterOOMWatchOptions{Target: sdkresource.TiCDC})
	if err != nil {
		return err
	}
	tikvEventChan, err := tc.WatchOOM(cancelCtx, sdkresource.TiDBClusterOOMWatchOptions{Target: sdkresource.TiKV})
	if err != nil {
		return err
	}
	cdcOOMCount := 0
	tikvOOMCount := 0
	go func() {
		for {
			var oom sdkresource.OOMEvent
			select {
			case oom = <-cdcEventChan:
				log.Warn("ticdc is oom", zap.String("node", oom.InstanceName))
				cdcOOMCount++
			case <-cancelCtx.Done():
				return
			}
		}
	}()
	go func() {
		for {
			var oom sdkresource.OOMEvent
			select {
			case oom = <-tikvEventChan:
				log.Warn("tikv is oom", zap.String("node", oom.InstanceName))
				tikvOOMCount++
			case <-cancelCtx.Done():
				return
			}
		}
	}()
	log.Info("watch oom finished", zap.Int("cdcOOMCount", cdcOOMCount), zap.Int("tikvOOMCount", tikvOOMCount))
	return nil
}