package steps

import (
	"context"
	"fmt"
	"time"

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

type kafkaControllerFailure struct {
	handle   *apiresource.ResourceHandle
	duration time.Duration
}

func NewKafkaControllerFailure(handle *apiresource.ResourceHandle, duration time.Duration) *kafkaControllerFailure {
	return &kafkaControllerFailure{
		handle:   handle,
		duration: duration,
	}
}

func (k *kafkaControllerFailure) Execute(ctx context.Context) error {
	return newKafkaBrokerFailure(k.handle, resource.KafkaClusterSelectorOpts{Controller: true}, k.duration).Execute(ctx)
}

func (k *kafkaControllerFailure) String() string {
	return fmt.Sprintf("kafkaControllerFailure: kafkaHandler=%+v, duration=%+v", k.handle, k.duration)
}

func NewKafkaRandomBrokerFailure(handle *apiresource.ResourceHandle, duration time.Duration) *kafkaRandomBrokerFailure {
	return &kafkaRandomBrokerFailure{
		handle:   handle,
		duration: duration,
	}
}

func (k *kafkaRandomBrokerFailure) Execute(ctx context.Context) error {
	return newKafkaBrokerFailure(k.handle, resource.KafkaClusterSelectorOpts{RandomSameOne: true}, k.duration).Execute(ctx)
}

func (k *kafkaRandomBrokerFailure) String() string {
	return fmt.Sprintf("kafkaRandomBrokerFailure: kafkaHandler=%+v, duration=%+v", k.handle, k.duration)
}

type kafkaRandomBrokerFailure struct {
	handle   *apiresource.ResourceHandle
	duration time.Duration
}

type kafkaBrokerFailure struct {
	handle   *apiresource.ResourceHandle
	duration time.Duration

	selectOption resource.KafkaClusterSelectorOpts
}

func newKafkaBrokerFailure(handle *apiresource.ResourceHandle, selectOption resource.KafkaClusterSelectorOpts, duration time.Duration) *kafkaBrokerFailure {
	return &kafkaBrokerFailure{
		handle:       handle,
		duration:     duration,
		selectOption: selectOption,
	}
}

func (s *kafkaBrokerFailure) String() string {
	return fmt.Sprintf("kafkaBrokerFailure: kafkaHandler=%+v, selectOption=%+v, duration=%+v", s.handle, s.selectOption, s.duration)
}

func (s *kafkaBrokerFailure) Execute(ctx context.Context) error {
	return newKafkaChaosTask(s.handle, nil,
		s.selectOption, nil,
		&resource.FailureExperimentSpec{},
		s.duration).Execute(ctx)
}

type kafkaChaosTask struct {
	kc *apiresource.ResourceHandle
	tc *apiresource.ResourceHandle

	// how long the chaos will exist after the creation.
	duration time.Duration

	spec         resource.ExperimentSpec
	selectOption resource.KafkaClusterSelectorOpts
	// some chaos may need a target, for kafka chaos, only target to `TiDB` should be enough
	// todo: add a interface for all `SelectorOpts`.
	targetOption *resource.TiDBClusterSelectorOpts
}

func newKafkaChaosTask(kafkaHandler, tidbHandler *apiresource.ResourceHandle, selectOption resource.KafkaClusterSelectorOpts,
	targetOption *resource.TiDBClusterSelectorOpts, spec resource.ExperimentSpec, duration time.Duration) *kafkaChaosTask {
	return &kafkaChaosTask{
		kc:           kafkaHandler,
		tc:           tidbHandler,
		selectOption: selectOption,
		targetOption: targetOption,
		duration:     duration,
		spec:         spec,
	}
}

func (s *kafkaChaosTask) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(s.kc)
	if err != nil {
		return errors.Trace(err)
	}
	kc, ok := resolved.(resource.KafkaCluster)
	if !ok {
		return fmt.Errorf("kafka chaos task cannot find kafka cluster, but: %+v", resolved)
	}

	selector, err := kc.ChaosSelect(s.selectOption)
	if err != nil {
		return errors.Trace(err)
	}

	chaosOption := resource.InjectChaosOptions{
		Selector:       selector,
		ExperimentSpec: s.spec,
	}

	if s.targetOption != nil {
		if s.tc == nil {
			return fmt.Errorf("tidb cluster handler not found")
		}
		resolved, err = core.ResolveResourceHandle(s.tc)
		if err != nil {
			return errors.Trace(err)
		}
		tc, ok := resolved.(resource.TiDBCluster)
		if !ok {
			return fmt.Errorf("kafka chaos task with target should be TiDBCluster, but got: %+v", resolved)
		}

		target, err := tc.ChaosSelect(*s.targetOption)
		if err != nil {
			return errors.Trace(err)
		}
		chaosOption.Target = &target
	}

	chaosID, err := kc.ChaosClient().Create(chaosOption)
	if err != nil {
		log.Info("create chaos failed", zap.Error(err), zap.Reflect("chaosOption", chaosOption))
		return errors.Trace(err)
	}
	defer func() {
		if err := kc.ChaosClient().Delete(chaosID); err != nil {
			log.Warn("delete chaos failed", zap.Error(err), zap.Reflect("chaosOption", chaosOption))
		}
	}()

	if chaosID == nil {
		log.Info("create chaos failed", zap.Any("chaosOption", chaosOption))
		return fmt.Errorf("create chaos failed, option=%+v", chaosOption)
	}

	time.Sleep(s.duration)

	log.Info("run chaos finished")

	return nil
}

func (s *kafkaChaosTask) String() string {
	return fmt.Sprintf("kafkaChaosTask: spec=%+v, duration=%+v", s.spec, s.duration)
}

type kafkaNetworkLossChaosTask struct {
	handle       *apiresource.ResourceHandle
	selectOption resource.KafkaClusterSelectorOpts

	duration    time.Duration
	loss        string
	correlation string
}

func newKafkaNetworkLossChaosTask(handle *apiresource.ResourceHandle, selectOption resource.KafkaClusterSelectorOpts, duration time.Duration) *kafkaNetworkLossChaosTask {
	return &kafkaNetworkLossChaosTask{
		handle:       handle,
		selectOption: selectOption,
		duration:     duration,
		loss:         "100",
		correlation:  "25",
	}
}

func (s *kafkaNetworkLossChaosTask) String() string {
	return fmt.Sprintf("kafkaNetworkLossChaosTask: kafkaHandler=%+v, selectOption=%+v, duration=%+v, loss=%+v, correlation=%+v",
		s.handle, s.selectOption, s.duration, s.loss, s.correlation)
}

func (s *kafkaNetworkLossChaosTask) Execute(ctx context.Context) error {
	return newKafkaChaosTask(s.handle, nil,
		s.selectOption, nil,
		&resource.NetworkLossSpec{Loss: s.loss, Correlation: s.correlation},
		s.duration).Execute(ctx)
}

func NewRandomKafkaNetworkLoss(handle *apiresource.ResourceHandle, duration time.Duration) *kafkaNetworkLossChaosTask {
	return newKafkaNetworkLossChaosTask(handle,
		resource.KafkaClusterSelectorOpts{RandomSameOne: true}, duration)
}

func NewKafkaControllerNetworkLoss(handle *apiresource.ResourceHandle, duration time.Duration) *kafkaNetworkLossChaosTask {
	return newKafkaNetworkLossChaosTask(handle,
		resource.KafkaClusterSelectorOpts{Controller: true}, duration)
}

func NewAllKafkaBrokersNetworkLoss(handle *apiresource.ResourceHandle, duration time.Duration) *kafkaNetworkLossChaosTask {
	return newKafkaNetworkLossChaosTask(handle,
		resource.KafkaClusterSelectorOpts{All: true}, duration)
}

type kafka2CDCNetworkPartitionChaosTask struct {
	kafkaHandler *apiresource.ResourceHandle
	tidbHandler  *apiresource.ResourceHandle

	selectOption resource.KafkaClusterSelectorOpts
	targetOption *resource.TiDBClusterSelectorOpts

	duration time.Duration
}

func newKafkaNetworkPartitionChaosTask(source, target *apiresource.ResourceHandle,
	selectOption resource.KafkaClusterSelectorOpts, targetOption *resource.TiDBClusterSelectorOpts, duration time.Duration) *kafka2CDCNetworkPartitionChaosTask {
	return &kafka2CDCNetworkPartitionChaosTask{
		kafkaHandler: source,
		tidbHandler:  target,
		selectOption: selectOption,
		targetOption: targetOption,
		duration:     duration,
	}
}

func (s *kafka2CDCNetworkPartitionChaosTask) String() string {
	return fmt.Sprintf("kafka2CDCNetworkPartitionChaosTask: kafkaHandler=%+v, selectOption=%+v, duration=%+v", s.kafkaHandler, s.selectOption, s.duration)
}

func (s *kafka2CDCNetworkPartitionChaosTask) Execute(ctx context.Context) error {
	return newKafkaChaosTask(s.kafkaHandler, s.tidbHandler,
		s.selectOption, s.targetOption,
		&resource.NetworkPartitionSpec{
			Direction: "to",
		}, s.duration).Execute(ctx)
}

func NewKafkaController2CDCOwnerNetworkPartition(handle, target *apiresource.ResourceHandle, duration time.Duration) *kafka2CDCNetworkPartitionChaosTask {
	return newKafkaNetworkPartitionChaosTask(handle, target,
		resource.KafkaClusterSelectorOpts{Controller: true},
		&resource.TiDBClusterSelectorOpts{
			TiDBClusterCDCSelectorSpec: &resource.TiDBClusterCDCSelectorSpec{
				IsOwner: &[]bool{true, false}[0],
			},
		},
		duration)
}

func NewRandomKafka2CDCOwnerNetworkPartition(handle, target *apiresource.ResourceHandle, duration time.Duration) *kafka2CDCNetworkPartitionChaosTask {
	return newKafkaNetworkPartitionChaosTask(handle, target,
		resource.KafkaClusterSelectorOpts{RandomSameOne: true},
		&resource.TiDBClusterSelectorOpts{
			TiDBClusterCDCSelectorSpec: &resource.TiDBClusterCDCSelectorSpec{
				IsOwner: &[]bool{true, false}[0],
			},
		},
		duration)
}

func NewAllKafka2CDCOwnerNetworkPartition(handle, target *apiresource.ResourceHandle, duration time.Duration) *kafka2CDCNetworkPartitionChaosTask {
	return newKafkaNetworkPartitionChaosTask(handle, target,
		resource.KafkaClusterSelectorOpts{All: true},
		&resource.TiDBClusterSelectorOpts{
			TiDBClusterCDCSelectorSpec: &resource.TiDBClusterCDCSelectorSpec{
				IsOwner: &[]bool{true, false}[0],
			},
		},
		duration)
}

func NewKafkaController2RandomCDCNetworkPartition(handle, target *apiresource.ResourceHandle, duration time.Duration) *kafka2CDCNetworkPartitionChaosTask {
	return newKafkaNetworkPartitionChaosTask(handle, target,
		resource.KafkaClusterSelectorOpts{Controller: true},
		&resource.TiDBClusterSelectorOpts{
			TiDBClusterCDCSelectorSpec: &resource.TiDBClusterCDCSelectorSpec{
				RandomSameOne: true,
			},
		},
		duration)
}

func NewRandomKafka2RandomCDCNetworkPartition(handle, target *apiresource.ResourceHandle, duration time.Duration) *kafka2CDCNetworkPartitionChaosTask {
	return newKafkaNetworkPartitionChaosTask(handle, target,
		resource.KafkaClusterSelectorOpts{RandomSameOne: true},
		&resource.TiDBClusterSelectorOpts{
			TiDBClusterCDCSelectorSpec: &resource.TiDBClusterCDCSelectorSpec{
				RandomSameOne: true,
			},
		},
		duration)
}

func NewAllKafka2RandomCDCNetworkPartition(handle, target *apiresource.ResourceHandle, duration time.Duration) *kafka2CDCNetworkPartitionChaosTask {
	return newKafkaNetworkPartitionChaosTask(handle, target,
		resource.KafkaClusterSelectorOpts{All: true},
		&resource.TiDBClusterSelectorOpts{
			TiDBClusterCDCSelectorSpec: &resource.TiDBClusterCDCSelectorSpec{
				RandomSameOne: true,
			},
		},
		duration)
}

type cdc2KafkaNetworkPartitionChaosTask struct {
	kafkaHandler *apiresource.ResourceHandle
	tidbHandler  *apiresource.ResourceHandle

	selectOption *resource.TiDBClusterSelectorOpts
	targetOption resource.KafkaClusterSelectorOpts

	duration time.Duration
}

func (s *cdc2KafkaNetworkPartitionChaosTask) String() string {
	return fmt.Sprintf("CDC2KafkaNetworkPartitionChaosTask: tidbHandler=%+v, selectOption=%+v, duration=%+v", s.tidbHandler, s.selectOption, s.duration)
}

func (s *cdc2KafkaNetworkPartitionChaosTask) Execute(ctx context.Context) error {
	return newKafkaChaosTask(s.kafkaHandler, s.tidbHandler,
		s.targetOption, s.selectOption,
		&resource.NetworkPartitionSpec{
			Direction: "from",
		}, s.duration).Execute(ctx)
}

func newCDC2KafkaNetworkPartitionChaosTask(target, source *apiresource.ResourceHandle,
	selectOption *resource.TiDBClusterSelectorOpts, targetOption resource.KafkaClusterSelectorOpts, duration time.Duration) *cdc2KafkaNetworkPartitionChaosTask {
	return &cdc2KafkaNetworkPartitionChaosTask{
		kafkaHandler: target,
		tidbHandler:  source,
		selectOption: selectOption,
		targetOption: targetOption,
		duration:     duration,
	}
}

func NewCDCOwner2KafkaControllerNetworkPartition(target, source *apiresource.ResourceHandle, duration time.Duration) *cdc2KafkaNetworkPartitionChaosTask {
	return newCDC2KafkaNetworkPartitionChaosTask(target, source,
		&resource.TiDBClusterSelectorOpts{
			TiDBClusterCDCSelectorSpec: &resource.TiDBClusterCDCSelectorSpec{
				IsOwner: &[]bool{true, false}[0],
			},
		},
		resource.KafkaClusterSelectorOpts{Controller: true},
		duration)
}