package steps

import (
	"context"
	"math/rand"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/test-infra/caselib/pkg/utils"
	"github.com/pingcap/tiflow/dm/dm/pb"
	"go.uber.org/zap"

	"github.com/pingcap/test-infra/caselib/pkg/host"
	dmmodel "github.com/pingcap/test-infra/caselib/pkg/model/dm"
	apiresource "github.com/pingcap/test-infra/common/model/resource"
)

type operateSourceStep struct {
	dmHost     *host.DMHost
	sourcePath string
	op         dmmodel.OperateSourceType
}

func (cs *operateSourceStep) Execute(ctx context.Context) error {
	err := cs.dmHost.OperateSource(&dmmodel.SourceOption{
		ConfigPath:  cs.sourcePath,
		OperateType: cs.op,
	})
	if err != nil {
		return err
	}
	return nil
}

func (cs *operateSourceStep) String() string {
	return cs.op.String() + " source by source files: " + cs.sourcePath
}

func (b *Builder) DmCreateSources(dm *apiresource.ResourceHandle, dmHost *host.DMHost, sourcePaths []string) *Builder {
	for _, path := range sourcePaths {
		b.DmCreateSource(dm, dmHost, path)
	}
	return b
}

func (b *Builder) DmCreateSource(dm *apiresource.ResourceHandle, dmHost *host.DMHost, sourcePath string) *Builder {
	b.steps = append(b.steps, &operateSourceStep{
		dmHost:     dmHost,
		sourcePath: sourcePath,
		op:         dmmodel.Create,
	})
	return b
}

func (b *Builder) DmStopSource(dm *apiresource.ResourceHandle, dmHost *host.DMHost, sourcePath string) *Builder {
	b.steps = append(b.steps, &operateSourceStep{
		dmHost:     dmHost,
		sourcePath: sourcePath,
		op:         dmmodel.Stop,
	})
	return b
}

func (b *Builder) DmShowSource(dm *apiresource.ResourceHandle, dmHost *host.DMHost, sourcePath string) *Builder {
	b.steps = append(b.steps, &operateSourceStep{
		dmHost:     dmHost,
		sourcePath: sourcePath,
		op:         dmmodel.Show,
	})
	return b
}

type killSourceWorkerStep struct {
	dmHost     *host.DMHost
	sourceName string
}

// NewKillSourceWorkerStep create a killSourceWorkerStep which kills the worker bound to the source
func NewKillSourceWorkerStep(dmHost *host.DMHost, sourceName string) *killSourceWorkerStep {
	return &killSourceWorkerStep{
		dmHost:     dmHost,
		sourceName: sourceName,
	}
}

func (k killSourceWorkerStep) String() string {
	return "kill worker of " + k.sourceName
}

func (k killSourceWorkerStep) Execute(ctx context.Context) error {
	_, status, err := k.dmHost.QueryStatus("", k.sourceName)
	if err != nil {
		return err
	}
	if len(status.Sources) != 1 || status.Sources[0].SourceStatus == nil {
		return errors.Errorf("failed to get source worker", zap.Reflect("status", status))
	}
	if _, err := k.dmHost.KillWorker(status.Sources[0].SourceStatus.Worker); err != nil {
		return err
	}
	return nil
}

type killLeadAndRandomMaster struct {
	dmHost *host.DMHost
}

// NewKillLeadAndRandomMaster create a killLeadAndRandomMaster which kills the master leader and random master
// right now we use master-0 to do dmctl command, if we kill the leader, the list-member may fail
// so kill leader and random master is merged into one step
func NewKillLeadAndRandomMaster(dmHost *host.DMHost) *killLeadAndRandomMaster {
	return &killLeadAndRandomMaster{
		dmHost: dmHost,
	}
}

func (k killLeadAndRandomMaster) String() string {
	return "kill lead and random master"
}

func (k killLeadAndRandomMaster) Execute(ctx context.Context) error {
	var leader string
	{
		resp, err := k.dmHost.ListMember("-l")
		if err != nil {
			return err
		}
		if !resp.Result || len(resp.Members) != 1 || resp.Members[0].GetLeader() == nil {
			return errors.Errorf("failed to get leader", zap.Reflect("status", resp))
		}
		leader = resp.Members[0].GetLeader().Name
	}
	var otherMasters []string
	{
		resp, err := k.dmHost.ListMember("-m")
		if err != nil {
			return err
		}
		if !resp.Result || len(resp.Members) != 1 || resp.Members[0].GetMaster() == nil {
			return errors.Errorf("failed to get masters", zap.Reflect("status", resp))
		}
		for _, m := range resp.Members[0].GetMaster().Masters {
			if m.Name != leader {
				otherMasters = append(otherMasters, m.Name)
			}
		}
	}

	if _, err := k.dmHost.KillMaster(leader); err != nil {
		return err
	}

	time.Sleep(1 * time.Second)

	if _, err := k.dmHost.KillMaster(otherMasters[rand.Intn(len(otherMasters))]); err != nil {
		return err
	}

	return nil
}

type checkIfSourceRelayStartStep struct {
	dmHost     *host.DMHost
	sourceName string
}

// getSourceWorkerStep create a getSourceWorkerStep which get the worker bound to the source
func NewCheckIFSourceRelayStartStep(dmHost *host.DMHost, sourceName string) *checkIfSourceRelayStartStep {
	return &checkIfSourceRelayStartStep{
		dmHost:     dmHost,
		sourceName: sourceName,
	}
}

func (k checkIfSourceRelayStartStep) String() string {
	return "get source relay status for " + k.sourceName
}

func (k checkIfSourceRelayStartStep) Execute(ctx context.Context) error {
	_, status, err := k.dmHost.QueryStatus("", k.sourceName)
	if err != nil {
		return err
	}
	if len(status.Sources) != 1 || status.Sources[0].SourceStatus == nil {
		return errors.Errorf("failed to get source worker", zap.Reflect("status", status))
	}
	// TODO: how to start relay for all sources?
	// for _, source := range status.Sources {
	// 	err = k.dmHost.StartRelay(k.sourceName, source.SourceStatus.Worker)
	// 	if err != nil {
	// 		return errors.Errorf("failed to start relay")
	// 	}
	// }
	if status.Sources[0].SourceStatus.RelayStatus == nil {
		return errors.Errorf("failed to start relay for source", zap.Reflect("status", status))
	}

	return nil

}

type checkIfSourceRelayStopStep struct {
	dmHost     *host.DMHost
	sourceName string
}

// getSourceWorkerStep create a getSourceWorkerStep which get the worker bound to the source
func NewCheckIfSourceRelayStopStep(dmHost *host.DMHost, sourceName string) *checkIfSourceRelayStopStep {
	return &checkIfSourceRelayStopStep{
		dmHost:     dmHost,
		sourceName: sourceName,
	}
}

func (k checkIfSourceRelayStopStep) String() string {
	return "get source relay status for " + k.sourceName
}

func (k checkIfSourceRelayStopStep) Execute(ctx context.Context) error {
	_, status, err := k.dmHost.QueryStatus("", k.sourceName)
	if err != nil {
		return err
	}
	if len(status.Sources) != 1 || status.Sources[0].SourceStatus == nil {
		return errors.Errorf("failed to get source worker", zap.Reflect("status", status))
	}
	if status.Sources[0].SourceStatus.RelayStatus != nil {
		return errors.Errorf("failed to stop relay for source", zap.Reflect("status", status))
	}

	return nil

}

type transferSource struct {
	dmHost     *host.DMHost
	sourceName string
}

func (t transferSource) Execute(ctx context.Context) error {
	var workerName string
	response := utils.Try(t.dmHost.ListMember("--worker")).(*pb.ListMemberResponse)
	if !response.Result || len(response.Members) != 1 || response.Members[0].GetWorker() == nil {
		return errors.Errorf("failed to get worker", zap.Reflect("status", response))
	}
	for _, workerInfo := range response.Members[0].GetWorker().Workers {
		if workerInfo.Stage == "free" {
			workerName = workerInfo.Name
			break
		}
	}
	if workerName == "" {
		return errors.Errorf("failed to get free worker", zap.Reflect("status", response))
	}
	return t.dmHost.TransferSource(t.sourceName, workerName)
}

func (t transferSource) String() string {
	return "transfer-source for " + t.sourceName
}

func (b *Builder) DmTransferSource(dmHost *host.DMHost, sourceName string) *Builder {
	b.steps = append(b.steps, &transferSource{
		dmHost, sourceName,
	})
	return b
}

type transferSourceToWorker struct {
	dmHost     *host.DMHost
	sourceName string
	worker     string
}

func (t transferSourceToWorker) Execute(ctx context.Context) error {
	return t.dmHost.TransferSource(t.sourceName, t.worker)
}

func (t transferSourceToWorker) String() string {
	return "transfer source" + t.sourceName + "to worker" + t.worker
}

func (b *Builder) DmTransferSourceToWorker(dmHost *host.DMHost, sourceName string, worker string) *Builder {
	b.steps = append(b.steps, &transferSourceToWorker{
		dmHost, sourceName, worker,
	})
	return b
}

type getSourceWorker struct {
	dmHost     *host.DMHost
	sourceName string
	worker     *string
}

func (k *getSourceWorker) Execute(ctx context.Context) error {
	_, status, err := k.dmHost.QueryStatus("", k.sourceName)
	if err != nil {
		return err
	}
	if len(status.Sources) != 1 || status.Sources[0].SourceStatus == nil {
		return errors.Errorf("failed to get source worker", zap.Reflect("status", status))
	}
	*k.worker = status.Sources[0].SourceStatus.Worker
	return nil
}

func (k *getSourceWorker) String() string {
	return "get source worker for " + k.sourceName
}

func (b *Builder) DmGetSourceWorker(dmHost *host.DMHost, sourceName string, worker *string) *Builder {
	b.steps = append(b.steps, &getSourceWorker{
		dmHost, sourceName, worker,
	})
	return b
}

func NewGetDmSourceWorker(dmHost *host.DMHost, sourceName string, worker *string) *getSourceWorker {
	return &getSourceWorker{
		dmHost, sourceName, worker,
	}
}

// Tiup create source
type tiupDmCreateSource struct {
	dmHost     *host.DMHost
	sourcePath string
	// sourceNameList []string
}

func (w tiupDmCreateSource) String() string {
	return "tiupDmCreateSource " + w.sourcePath
}

func (w tiupDmCreateSource) Execute(ctx context.Context) error {
	// wait at most for 5 minutes, if the task is still not synced, return error
	tmpPath := utils.TempPath(w.sourcePath)
	// err := w.dmHost.TiupCreateSource(tmpPath)
	// if err != nil {
	// 	return err
	// }
	utils.Try(w.dmHost.TiupCreateSource(tmpPath))
	return nil
}

func NewTiupDmCreateSource(dmHost *host.DMHost, sourcePath string) *tiupDmCreateSource {
	return &tiupDmCreateSource{
		dmHost:     dmHost,
		sourcePath: sourcePath,
	}
}

func (b *Builder) TiupDmCreateSources(dm *apiresource.ResourceHandle, dmHost *host.DMHost, sourcePaths []string) *Builder {
	for _, sourcePath := range sourcePaths {
		b.TiupDmCreateSource(dm, dmHost, sourcePath)
	}
	return b
}

func (b *Builder) TiupDmCreateSource(dm *apiresource.ResourceHandle, dmHost *host.DMHost, sourcePath string) *Builder {
	b.steps = append(b.steps, NewTiupDmCreateSource(dmHost, sourcePath))
	return b
}