package steps

import (
	"time"

	"github.com/pingcap/test-infra/caselib/pkg/model/kafka"
	"github.com/pingcap/test-infra/caselib/pkg/model/ticdc"
	"github.com/pingcap/test-infra/caselib/pkg/model/workload"
	"github.com/pingcap/test-infra/common/model/resource"

	sdkresource "github.com/pingcap/test-infra/sdk/resource"
)

// Builder is used to build testcase.
type Builder struct {
	steps []Step
}

// NewBuilder returns a Builder instance.
func NewBuilder() *Builder {
	return &Builder{}
}

// Include applies the callback over the builder.
// This allowes users to define functions of theirselves and insert them into the builder chain:
// builder.
//	Include(SomeUserDefinedStep(arg1, arg2))
func (b *Builder) Include(f func(b *Builder)) *Builder {
	f(b)
	return b
}

// Serial appends the steps to the tail of queue.
func (b *Builder) Serial(steps ...Step) *Builder {
	if len(steps) > 0 {
		b.steps = append(b.steps, steps...)
	}

	return b
}

// Parallel appends a parallel step to the current step collection.
func (b *Builder) Parallel(steps ...Step) *Builder {
	if len(steps) > 0 {
		b.steps = append(b.steps, &Parallel{inner: steps})
	}
	return b
}

func (b *Builder) LoadEnv() *Builder {
	b.steps = append(b.steps, NewLoadEnv())
	return b
}

func (b *Builder) LoadResourceFromEnv(envName string, target interface{}) *Builder {
	b.steps = append(b.steps, NewLoadResourceFromEnv(envName, target))
	return b
}

func (b *Builder) ScaleTiDBCluster(tp sdkresource.TiDBCompType, replica uint32, timeout time.Duration) *Builder {
	b.steps = append(b.steps, NewScaleTiDBCluster(tp, replica, timeout))
	return b
}

// scale tidb cluster after Init tidb cluster
// Board.SetTiDBCluster() before using
func (b *Builder) ScaleTiDBcomponent(resourceHandleTiDB *resource.ResourceHandle, tp sdkresource.TiDBCompType, replica uint32, timeout time.Duration) *Builder {
	b.steps = append(b.steps, NEWScaleTiDBcomponent(resourceHandleTiDB, tp, replica, timeout))
	return b
}

func (b *Builder) UpgradeTiDBCluster(tp sdkresource.TiDBCompType, version *string, timeout time.Duration) *Builder {
	b.steps = append(b.steps, NewUpgradeTiDBCluster(tp, version, timeout))
	return b
}

// Upgrade tidb cluster after Init tidb cluster
// Board.SetTiDBCluster() before using
func (b *Builder) UpgradeTiDBAllComponent(resourceHandleTiDB *resource.ResourceHandle, version *string, timeout time.Duration) *Builder {
	b.steps = append(b.steps, NewUpgradeTiDBAllComponent(resourceHandleTiDB, version, timeout))
	return b
}

func (b *Builder) UpdateConfig(tp sdkresource.TiDBCompType, version string, timeout time.Duration) *Builder {
	b.steps = append(b.steps, NewUpdateConfig(tp, version, timeout))
	return b
}

func (b *Builder) InitCDCComponents(resourceHandleTiDB, resourceHandleSink *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewInitCDCComponents(resourceHandleTiDB, resourceHandleSink))
	return b
}

func (b *Builder) RemoveCDCComponents(resourceHandleTiDB, resourceHandleSink *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewRemoveCDCComponents())
	return b
}

func (b *Builder) InitKafka(resourceHandleSink *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewInitKafka(resourceHandleSink))
	return b
}

func (b *Builder) RunKafka(option kafka.TopicOption) *Builder {
	b.steps = append(b.steps, NewRunKafka(option))
	return b
}

func (b *Builder) RunKafkaConsumer(broker, consumer, consumerTarget *resource.ResourceHandle, option *kafka.ConsumerOption) *Builder {
	b.steps = append(b.steps, NewRunKafkaConsumer(broker, consumer, consumerTarget, option))
	return b
}

func (b *Builder) KillKafkaConsumer(consumer *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewKillKafkaConsumer(consumer))
	return b
}

func (b *Builder) DeleteKafkaTopic(broker *resource.ResourceHandle, topic string) *Builder {
	b.steps = append(b.steps, NewDeleteKafkaTopic(broker, topic))
	return b
}

func (b *Builder) InitWorkload(resourceHandleWorkload *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewInitWorkload(resourceHandleWorkload))
	return b
}

// RunWorkload will run workload in a new goroutine, `WaitWorkloadFinish` must be called in next steps.
func (b *Builder) RunWorkload(option workload.Option) *Builder {
	b.steps = append(b.steps, NewRunWorkload(option))
	return b
}

func (b *Builder) InitTiup(resourceHandleTiUP, resourceHandleSink *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewInitTiup(resourceHandleTiUP, resourceHandleSink))
	return b
}

func (b *Builder) DeployTiupClusterPDDown(option workload.TiupClusterOption) *Builder {
	b.steps = append(b.steps, NewDeployTiupClusterPDDown(option))
	return b
}

func (b *Builder) DeployTiupCluster(option workload.TiupClusterOption) *Builder {
	b.steps = append(b.steps, NewDeployTiupCluster(option))
	return b
}

func (b *Builder) DepstroyTiupCluster(option workload.TiupClusterOption) *Builder {
	b.steps = append(b.steps, NewDestroyTiupCluster(option))
	return b
}

func (b *Builder) DeployBinlogTiupCluster(option workload.TiupClusterOption, workload *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewDeployBinlogTiupCluster(option, workload))
	return b
}

func (b *Builder) ScaleOutTiupBinlogCluster(option workload.TiupClusterOption, workload *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewScaleOutTiupBinlogCluster(workload, option))
	return b
}

func (b *Builder) ScaleInTiupBinlogCluster(option workload.TiupClusterOption, workload *resource.ResourceHandle, node string) *Builder {
	b.steps = append(b.steps, NewScaleInTiupBinlogCluster(workload, option, node))
	return b
}

func (b *Builder) DeployTiupDmCluster(option workload.TiupDmOption, workload *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewDeployTiupDmCluster(workload, option))
	return b
}

func (b *Builder) UpgradeTiupDmCluster(option workload.TiupDmOption, workload *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NEWUpgradeTiupDmCluster(workload, option))
	return b
}

func (b *Builder) ScaleOutTiupDmCluster(option workload.TiupDmOption, workload *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewScaleOutTiupDmCluster(workload, option))
	return b
}

func (b *Builder) ScaleInTiupDmCluster(option workload.TiupDmOption, workload *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewScaleInTiupDmCluster(workload, option))
	return b
}

func (b *Builder) TiupCreateChangefeed(option workload.TiupChangefeedOption) *Builder {
	b.steps = append(b.steps, NewTiupCreateChangefeed(option))
	return b
}

func (b *Builder) RunWorkloadTiup(workloadOption workload.Option) *Builder {
	b.steps = append(b.steps, NewRunWorkloadTiup(workloadOption))
	return b
}

func (b *Builder) ScaleOutTiCDCClusterTiup(option workload.TiupClusterOption) *Builder {
	b.steps = append(b.steps, NewScaleOutTiCDCClusterTiup(option))
	return b
}

func (b *Builder) RestartTiCDCTiup(option workload.TiupClusterOption) *Builder {
	b.steps = append(b.steps, NewRestartTiCDCTiup(option))
	return b
}

func (b *Builder) RunBinlogWorkloadTiup(workloadOption workload.Option, workload *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewRunBinlogWorkloadTiup(workloadOption, workload))
	return b
}

func (b *Builder) RunTiupMirrorSet(workloadOption workload.TiupMirrorOption) *Builder {
	b.steps = append(b.steps, NewRunTiupMirrorSet(workloadOption))
	return b
}

func (b *Builder) RunBinlogTiupMirrorSet(workloadOption workload.TiupMirrorOption, workload *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewRunBinlogTiupMirrorSet(workloadOption, workload))
	return b
}

func (b *Builder) RunTiupDmMirrorSet(workloadOption workload.TiupDmMirrorOption, workload *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewRunTiupDmMirrorSet(workloadOption, workload))
	return b
}

func (b *Builder) UpgradeTiupCluster(version *string) *Builder {
	b.steps = append(b.steps, NewUpgradeTiupCluster(version))
	return b
}

// OperateChangeFeed used to perform ticdc changefeed related work, such as create / query / list / stop / resume / remove
// it should be called after `InitCDCComponents`
func (b *Builder) OperateChangeFeed(flag ticdc.ChangeFeedParams) *Builder {
	b.steps = append(b.steps, NewChangeFeedTask(flag))
	return b
}

// OperateRedoLog will run redo apply or redo meta cmd
func (b *Builder) OperateRedoLog(flag ticdc.RedoParams) *Builder {
	b.steps = append(b.steps, NewRedo(flag))
	return b
}

// InitChangefeedConfig used to generate changefeed.toml
func (b *Builder) InitChangefeedConfig(changefeedCategory *string, changefeedContent *string) *Builder {
	b.steps = append(b.steps, NewChangeFeedConfig(changefeedCategory, changefeedContent))
	return b
}

func (b *Builder) CreateDatabase(handle *resource.ResourceHandle, dbName string) *Builder {
	b.steps = append(b.steps, NewCreateDatabase(handle, dbName))
	return b
}

func (b *Builder) CreateDatabaseTiup(dbName string) *Builder {
	b.steps = append(b.steps, NewCreateDatabaseTiup(dbName))
	return b
}

func (b *Builder) EnableTiDBGeneralLog(handler *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewEnableTiDBGeneralLog(handler))
	return b
}

func (b *Builder) SetGCLifeTime(handler *resource.ResourceHandle, gcLifeTime string) *Builder {
	b.steps = append(b.steps, NewGCLifeTime(handler, gcLifeTime))
	return b
}

func (b *Builder) SetMaxConnection(handler *resource.ResourceHandle, maxConnection int) *Builder {
	b.steps = append(b.steps, NewMaxConnection(handler, maxConnection))
	return b
}

// Check database available
func (b *Builder) CheckDatabaseAvailable(handle *resource.ResourceHandle, dbName string, filterTablename *string) *Builder {
	b.steps = append(b.steps, NewCheckDatabaseAvailable(handle, dbName, filterTablename))
	return b
}

// Execute sql statement
func (b *Builder) ExecuteSqlStatement(handle *resource.ResourceHandle, statements *string) *Builder {
	b.steps = append(b.steps, NewExecuteSqlStatement(handle, statements))
	return b
}

func (b *Builder) ExecuteSqlStatements(handle *resource.ResourceHandle, statements *string) *Builder {
	b.steps = append(b.steps, NewExecuteSqlStatements(handle, statements))
	return b
}

func (b *Builder) CheckConsistency(source, target *resource.ResourceHandle, dbName string) *Builder {
	b.steps = append(b.steps, NewCheckConsistency(source, target, dbName))
	return b
}

func (b *Builder) CheckUserDatabasesConsistency(source, target *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewCheckUserDatabasesConsistency(source, target))
	return b
}

func (b *Builder) CheckConsistencyTiup(tiupClusterOption workload.TiupClusterOption, dbName string) *Builder {
	b.steps = append(b.steps, NewCheckConsistencyTiup(tiupClusterOption, dbName))
	return b
}

func (b *Builder) CheckConsistencyBinlogTiup(tiupClusterOption workload.TiupClusterOption, downStream *resource.ResourceHandle, dbName string) *Builder {
	b.steps = append(b.steps, NewCheckConsistencyBinlogTiup(tiupClusterOption, downStream, dbName))
	return b
}

// Build returns a step that contains all steps appended by previous operation.
func (b *Builder) Build() Step {
	return &Serial{inner: b.steps}
}

func (b *Builder) Sleep(duration time.Duration) *Builder {
	b.steps = append(b.steps, NewSleepTask(duration))
	return b
}

// SignalWorkloadFinish will let other steps know that workload run finished, then can be executed.
func (b *Builder) SignalWorkloadFinish() *Builder {
	b.steps = append(b.steps, NewSignalWorkloadFinish())
	return b
}

// WaitWorkloadFinish must be called after `RunWorkload`
func (b *Builder) WaitWorkloadFinish() *Builder {
	b.steps = append(b.steps, NewWaitWorkloadFinish())
	return b
}

// SwitchOwner by inject pod failure chaos into cdc owner pod for every cron
func (b *Builder) SwitchOwner(target *resource.ResourceHandle, duration time.Duration) *Builder {
	b.steps = append(b.steps, NewSwitchOwner(target, duration))
	return b
}

func (b *Builder) MakeTiKVFailure(target *resource.ResourceHandle, duration time.Duration) *Builder {
	b.steps = append(b.steps, NewKVFailure(target, duration))
	return b
}

func (b *Builder) MakeTiKVFailureWithID(target *resource.ResourceHandle, failNode int, duration time.Duration) *Builder {
	b.steps = append(b.steps, NewKVFailureWithID(target, failNode, duration))
	return b
}

func (b *Builder) MakeTiDBFailure(target *resource.ResourceHandle, duration time.Duration) *Builder {
	b.steps = append(b.steps, NewTiDBFailure(target, duration))
	return b
}

func (b *Builder) MakeTiDBFailureWithID(target *resource.ResourceHandle, failNode int, duration time.Duration) *Builder {
	b.steps = append(b.steps, NewTiDBFailureWithID(target, failNode, duration))
	return b
}

func (b *Builder) SendFinishMark(dbName string) *Builder {
	b.steps = append(b.steps, NewSendFinishMark(dbName))
	return b
}

func (b *Builder) SendFinishMarkTiup(dbName string) *Builder {
	b.steps = append(b.steps, NewSendFinishMarkTiup(dbName))
	return b
}

func (b *Builder) WaitFinishMark(handle *resource.ResourceHandle, dbName string, timeout time.Duration) *Builder {
	b.steps = append(b.steps, NewWaitFinishMark(handle, dbName, timeout))
	return b
}

func (b *Builder) InjectAllTiKVNetworkLoss(target *resource.ResourceHandle, duration time.Duration) *Builder {
	b.steps = append(b.steps, NewAllTiKVNetworkLoss(target, duration))
	return b
}

func (b *Builder) PdSwitchLeader() *Builder {
	b.steps = append(b.steps, NewPdSwitchLeader())
	return b
}

// RunPeriodicalTask will run the task periodically, with the given cron
func (b *Builder) RunPeriodicalTask(lifeTime, cron time.Duration, task Step) *Builder {
	b.steps = append(b.steps, NewPeriodicalTask(lifeTime, cron, task))
	return b
}

func (b *Builder) RunLoopTask(lifeTime, cron time.Duration, roundRobin bool, task ...Step) *Builder {
	b.steps = append(b.steps, NewLoopTask(lifeTime, cron, roundRobin, task...))
	return b
}

func (b *Builder) RunKafkaRandomBrokerFailure(handle *resource.ResourceHandle, duration time.Duration) *Builder {
	b.steps = append(b.steps, NewKafkaRandomBrokerFailure(handle, duration))
	return b
}

func (b *Builder) InjectKafkaController2CDCOwnerNetworkPartition(handle, target *resource.ResourceHandle, duration time.Duration) *Builder {
	b.steps = append(b.steps, NewKafkaController2CDCOwnerNetworkPartition(handle, target, duration))
	return b
}

func (b *Builder) RepeatSteps(repeatTime int, repeatSteps ...Step) *Builder {
	for i := 0; i < repeatTime; i++ {
		b.steps = append(b.steps, repeatSteps...)
	}
	return b
}

func (b *Builder) WatchCDCTiKVOOM(handler *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewWatchCDCTiKVOOM(handler))
	return b
}

func (b *Builder) GetCDCVersion(handler *resource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewGetCDCVersion(handler))
	return b
}

func (b *Builder) GetCDCMetrics(handler *resource.ResourceHandle, startTime *time.Time) *Builder {
	b.steps = append(b.steps, NewGetCDCMetrics(handler, startTime))
	return b
}

func (b *Builder) GenerateCDCPerfSql(handler *resource.ResourceHandle, executeSql *string) *Builder {
	b.steps = append(b.steps, NewGenerateCDCPerfSql(handler, executeSql))
	return b
}

func (b *Builder) RecordTime(time *time.Time) *Builder {
	b.steps = append(b.steps, NewRecordTime(time))
	return b
}