package host

import (
	"fmt"
	"strings"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/caselib/pkg/command"
	"github.com/pingcap/test-infra/caselib/pkg/model/kafka"
	"github.com/pingcap/test-infra/caselib/pkg/model/ticdc"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
)

type TiCDCHost struct {
	Source resource.TiDBCluster
	Sink   resource.CDCSink

	// Peer indicates the index of the ticdc instance, always set to 0
	Peer int
}

type PerfTestbed struct {
	TiKVVersion string
	TiKVHashCommit string
	TiCDCVersion string
	TiCDCHashCommit string
	SinkURI string
	TestbedName string
}

var TestbedPara PerfTestbed

func NewTiCDCHost(source resource.TiDBCluster, sink resource.CDCSink) (*TiCDCHost, error) {
	cdcUrls, err := source.PeerURLs(resource.CDCAddr)
	if err != nil {
		logger.Panic("")
		return nil, errors.Trace(err)
	}
	if len(cdcUrls) <= 0 {
		return nil, errors.New("No valid TiCDC peer URLs")
	}

	return &TiCDCHost{
		Source: source,
		Sink:   sink,
		Peer:   0,
	}, nil
}

func (t *TiCDCHost) String() string {
	return fmt.Sprintf("TiCDCHost: Source=%+v, Sink=%+v, Peer=%+v", t.Source, t.Sink, t.Peer)
}

func (t *TiCDCHost) Commands() *command.CdcCmd {
	return &command.CdcCmd{ExecCmd: command.ExecCmd{Command: "/cdc", Options: nil}}
}

func (t *TiCDCHost) ChangeFeed(flags ticdc.ChangeFeedParams) (ticdc.Response, error) {
	commands, err := t.BuildCommands(flags)
	if err != nil {
		return nil, err
	}

	o, err := t.Exec(commands, 10*time.Second)
	if err == TimeoutErr {
		log.Error("run changefeed operation timeout")
		return nil, err
	}
	if err != nil {
		log.Error("run changefeed operation failed", zap.Any("output", o))
		return nil, err
	}
	if o.ExitCode != 0 {
		return nil, errors.Errorf("run changefeed operation failed, output=%+v", o)
	}

	if o.Stderr != "" {
		log.Warn("run changefeed operation", zap.String("stderr", o.Stderr))
	}

	result, err := flags.Unmarshal([]byte(o.Stdout))
	if err != nil {
		logger.Error("failed to unmarshal changefeed operation result", zap.String("stdout", o.Stdout), zap.Error(err))
		return nil, err
	}
	return result, nil
}

func (t *TiCDCHost) Redo(flags ticdc.RedoParams) (ticdc.Response, error) {
	commands, err := t.BuildRedoCommands(flags)
	if err != nil {
		return nil, err
	}

	o, err := t.Exec(commands, 100*time.Second)
	if err != nil {
		if err == TimeoutErr {
			log.Error("run redo operation timeout")
			return nil, err
		}
		log.Error("run redo operation failed", zap.Any("output", o))
		return nil, err
	}
	if o.ExitCode != 0 {
		return nil, errors.Errorf("run redo operation failed, output=%+v", o)
	}
	if o.Stderr != "" {
		log.Warn("run redo operation", zap.String("stderr", o.Stderr))
	}

	result, err := flags.Unmarshal([]byte(o.Stdout))
	if err != nil {
		logger.Error("failed to unmarshal redo operation result", zap.String("stdout", o.Stdout), zap.Error(err))
		return nil, err
	}
	return result, nil
}

func (t *TiCDCHost) BuildCommands(flags ticdc.ChangeFeedParams) (string, error) {
	args, err := t.Parse(flags)
	if err != nil {
		return "", err
	}

	commands := t.Commands().Cli().ChangeFeed()
	switch t := flags.(type) {
	case *ticdc.ChangeFeedCreateFlags:
		commands = commands.Create(args)
	case *ticdc.ChangeFeedUpdateFlags:
		commands = commands.Update(args)
	case *ticdc.ChangeFeedQueryFlags:
		commands = commands.Query(args)
	case *ticdc.ChangeFeedListFlags:
		commands = commands.List(args)
	case *ticdc.ChangeFeedAdminJobFlags:
		commands = commands.Admin(t.JobType, args)
	default:
		logger.Panic("invalid ticdc changefeed flags", zap.Reflect("flags", t))
	}

	return commands.Compose(), nil
}

func (t *TiCDCHost) BuildRedoCommands(flags ticdc.RedoParams) (string, error) {
	args, err := t.ParseRedo(flags)
	if err != nil {
		return "", err
	}

	commands := t.Commands().Redo()
	switch t := flags.(type) {
	case *ticdc.RedoApplyFlags:
		commands = commands.Apply(args)
	case *ticdc.RedoMetaFlags:
		commands = commands.Meta(args)
	default:
		logger.Panic("invalid ticdc redo flags", zap.Reflect("flags", t))
	}

	return commands.Compose(), nil
}

func (t *TiCDCHost) Parse(flags ticdc.ChangeFeedParams) (params []string, err error) {
	pdUrl, err := t.Source.ServiceURL(resource.PDAddr)
	if err != nil {
		return nil, err
	}
	params = append(params, "--pd="+pdUrl.String())

	if f, ok := flags.(*ticdc.ChangeFeedCreateFlags); ok {
		uri, err := t.Sink.SinkEndpoint()
		if err != nil {
			return nil, err
		}

		sinkURI := buildSinkURI(uri.String(), f.SinkType, f.Options)
		params = append(params, "--sink-uri="+sinkURI)
		TestbedPara.SinkURI = sinkURI
	}

	args, err := flags.Parse()
	if err != nil {
		return params, err
	}
	params = append(params, args...)
	return params, nil
}

func buildSinkURI(endpoint string, sinkType ticdc.SinkType, options map[string]string) string {
	if sinkType == ticdc.SinkTypeKafka {
		topicName, ok := options["topic"]
		if !ok {
			topicName = kafka.DefaultTopicName
		}
		delete(options, "topic")

		endpoint = endpoint + "/" + topicName
	}

	if len(options) != 0 {
		endpoint += "?"
		var p []string
		for k, v := range options {
			p = append(p, k+"="+v)
		}
		endpoint += strings.Join(p, "&")
	}
	return endpoint
}

func (t *TiCDCHost) ParseRedo(flags ticdc.RedoParams) (params []string, err error) {
	if _, ok := flags.(*ticdc.RedoApplyFlags); ok {
		uri, err := t.Sink.SinkEndpoint()
		if err != nil {
			return nil, err
		}

		s := uri.String()
		params = append(params, "--sink-uri="+s)
	}

	args, err := flags.Parse()
	if err != nil {
		return params, err
	}
	params = append(params, args...)
	return params, nil
}

func (t *TiCDCHost) Exec(cmd string, timeout time.Duration) (*command.ExecOutput, error) {
	done := make(chan *command.ExecOutput, 1)
	timeout = 1*time.Minute
	logger.Info("Execute command on ticdc", zap.String("command", cmd), zap.Duration("timeout", timeout), zap.Int("ticdc peer", t.Peer))
	go func() {
		stdout, stderr, exitCode, err := t.Source.Exec(resource.TiDBClusterExecOptions{Target: resource.TiCDC, Command: cmd, Peer: t.Peer})
		done <- &command.ExecOutput{Err: err, Stdout: stdout, Stderr: stderr, ExitCode: exitCode}
	}()

	select {
	case <-time.After(timeout):
		return nil, TimeoutErr
	case output := <-done:
		return output, errors.Trace(output.Err)
	}
}

func (t *TiCDCHost) TiCDCVersion() (*command.ExecOutput, error) {
	cmd := "/cdc version"
	logger.Info("Execute command on ticdc", zap.String("command", cmd), zap.Int("ticdc peer", t.Peer))
	stdout, stderr, exitCode, err := t.Source.Exec(resource.TiDBClusterExecOptions{Target: resource.TiCDC, Command: cmd, Peer: t.Peer})
	return &command.ExecOutput{Err: err, Stdout: stdout, Stderr: stderr, ExitCode: exitCode}, nil
}

func (t *TiCDCHost) TiKVVersion() (*command.ExecOutput, error) {
	cmd := "/tikv-server -V"
	stdout, stderr, exitCode, err := t.Source.Exec(resource.TiDBClusterExecOptions{Target: resource.TiKV, Command: cmd, Peer: t.Peer})

	return &command.ExecOutput{Err: err, Stdout: stdout, Stderr: stderr, ExitCode: exitCode}, nil
}