package steps

import (
	"context"
	"fmt"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/caselib/pkg/host"
	"github.com/pingcap/test-infra/caselib/pkg/model/kafka"
	"github.com/pingcap/test-infra/caselib/pkg/model/workload"
	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 initKafka struct {
	kafka *apiresource.ResourceHandle
}

func NewInitKafka(kafka *apiresource.ResourceHandle) Step {
	return &initKafka{
		kafka: kafka,
	}
}

func (s *initKafka) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(s.kafka)
	if err != nil {
		return err
	}
	return GetInner(ctx).Board().SetKafka(resolved.(resource.KafkaCluster))
}

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

type runKafka struct {
	option kafka.TopicOption
}

func NewRunKafka(option kafka.TopicOption) Step {
	return &runKafka{
		option: option,
	}
}

// Execute runKafka run the specified kafka create topic in a background goroutine.
func (s *runKafka) Execute(ctx context.Context) (err error) {
	command, err := s.option.Parse()
	if err != nil {
		return err
	}

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

	kafkaHost := host.NewKafkaHost(k)

	start := time.Now()

	if err := kafkaHost.Run(command); err != nil {
		log.Error("run kafka failed", zap.Error(err), zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
		return err
	}

	log.Info("run kafka finished", zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
	return nil
}

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

type runKafkaConsumer struct {
	broker   *apiresource.ResourceHandle
	consumer *apiresource.ResourceHandle
	target   *apiresource.ResourceHandle
	option   *kafka.ConsumerOption
}

func NewRunKafkaConsumer(broker, consumer, target *apiresource.ResourceHandle, option *kafka.ConsumerOption) Step {
	return &runKafkaConsumer{
		broker:   broker,
		consumer: consumer,
		target:   target,
		option:   option,
	}
}

func (r *runKafkaConsumer) String() string {
	return fmt.Sprintf("runKafkaConsumer: broker=%+v, consumer=%+v, target=%+v option=%+v",
		r.broker, r.consumer, r.target, r.option)
}

// Execute the kafka consumer, use `nohup` to make it run in background.
// todo: what would happen if the consumer pod dead
func (r *runKafkaConsumer) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(r.broker)
	if err != nil {
		return errors.Trace(err)
	}
	broker, ok := resolved.(resource.CDCSink)
	if !ok {
		return fmt.Errorf("consumer broker is not a valid cdc sink, broker: %+v", resolved)
	}

	endPoint, err := broker.SinkEndpoint()
	if err != nil {
		return errors.Trace(err)
	}

	resolved, err = core.ResolveResourceHandle(r.target)
	if err != nil {
		return errors.Trace(err)
	}
	mysql, ok := resolved.(resource.MySQL)
	if !ok {
		return fmt.Errorf("consumer target is not mysql, target: %+v", resolved)
	}
	endpoint, err := mysql.ServiceURL(resource.DBAddr)
	if err != nil {
		return errors.Trace(err)
	}

	command, err := r.option.WithBrokerAddr(endPoint.String()).WithTargetAddr(endpoint.String()).Parse()
	if err != nil {
		return errors.Trace(err)
	}

	resolved, err = core.ResolveResourceHandle(r.consumer)
	if err != nil {
		return errors.Trace(err)
	}
	consumer, ok := resolved.(resource.WorkloadNode)
	if !ok {
		return fmt.Errorf("consumer is not resource.WorkloadNode, but: %+v", resolved)
	}
	consumerHost := host.NewKafkaConsumerHost(consumer)

	start := time.Now()
	if err := consumerHost.Run(command); err != nil {
		log.Error("run kafka consumer failed", zap.Error(err), zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
	}
	log.Info("run kafka consumer finished", zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))

	return nil
}

type killKafkaConsumer struct {
	consumer *apiresource.ResourceHandle
}

func (k *killKafkaConsumer) String() string {
	return fmt.Sprintf("killKafkaConsumer: consumer=%+v", k.consumer)
}

func (k *killKafkaConsumer) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(k.consumer)
	if err != nil {
		return errors.Trace(err)
	}
	consumer, ok := resolved.(resource.WorkloadNode)
	if !ok {
		return fmt.Errorf("consumer is not resource.WorkloadNode, but: %+v", resolved)
	}
	consumerHost := host.NewKafkaConsumerHost(consumer)
	command := workload.NewCMD("pkill -9 -f cdc_kafka_consumer", 10*time.Second)
	if err := consumerHost.Run(command); err != nil {
		return errors.Trace(err)
	}

	return nil
}

func NewKillKafkaConsumer(consumer *apiresource.ResourceHandle) Step {
	return &killKafkaConsumer{
		consumer: consumer,
	}
}

type deleteKafkaTopic struct {
	broker *apiresource.ResourceHandle
	topic  string
}

func (d *deleteKafkaTopic) String() string {
	return fmt.Sprintf("deleteKafkaTopic: broker=%+v, topic=%+v", d.broker, d.topic)
}

func (d *deleteKafkaTopic) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(d.broker)
	if err != nil {
		return errors.Trace(err)
	}
	broker, ok := resolved.(resource.KafkaCluster)
	if !ok {
		return fmt.Errorf("broker is not a valid kafka cluster, broker: %+v", resolved)
	}
	endpoint, err := broker.ServiceURL()
	if err != nil {
		return errors.Trace(err)
	}

	command := kafka.NewCMD(fmt.Sprintf("sh /opt/kafka/bin/kafka-topics.sh --delete --bootstrap-server %s --topic %s", endpoint.String(), d.topic), time.Minute)

	kafkaHost := host.NewKafkaHost(broker)

	if err = kafkaHost.Run(command); err != nil {
		return errors.Trace(err)
	}

	return nil
}

func NewDeleteKafkaTopic(broker *apiresource.ResourceHandle, topic string) Step {
	return &deleteKafkaTopic{
		broker: broker,
		topic:  topic,
	}
}