package steps

import (
	"bytes"
	"context"
	"fmt"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/BurntSushi/toml"
	"github.com/oleiade/reflections"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/caselib/pkg/command"
	sdiconfig "github.com/pingcap/test-infra/caselib/pkg/model/dm"
	"github.com/pingcap/test-infra/caselib/pkg/utils"
	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 loadSyncDiffConfig struct {
	LocalConfigFile *string
	SyncDiffConfigs *sdiconfig.Config
}

func (l loadSyncDiffConfig) String() string {
	return fmt.Sprintf("load sync diff config file locally from %s", *l.LocalConfigFile)
}

func (l loadSyncDiffConfig) Execute(ctx context.Context) error {
	meta, err := toml.DecodeFile(*l.LocalConfigFile, l.SyncDiffConfigs)
	if err != nil {
		return errors.Trace(err)
	}
	if len(meta.Undecoded()) > 0 {
		return errors.Errorf("unknown keys in config file %v", meta.Undecoded())
	}
	return nil
}

// LoadSyncDiffConfig load sync diff configuration from local config files
func (b *Builder) LoadSyncDiffConfig(localSyncDiffConfig *string, syncDiffConfigs *sdiconfig.Config) *Builder {
	b.steps = append(b.steps, &loadSyncDiffConfig{localSyncDiffConfig, syncDiffConfigs})
	return b
}

func ExecSyncDiffCmd(syncDiff resource.WorkloadNode, cmd string, timeout time.Duration) (*command.ExecOutput, error) {
	done := make(chan *command.ExecOutput, 1)
	log.Info("Execute command on up sync diff", zap.String("command", cmd), zap.Duration("timeout", timeout))
	go func() {
		stdout, stderr, exitCode, err := syncDiff.Exec(resource.WorkloadNodeExecOptions{Command: cmd})
		done <- &command.ExecOutput{Err: err, Stdout: stdout, Stderr: stderr, ExitCode: exitCode}
	}()

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

type syncDiffInspectorCheck struct {
	syncDiff        *apiresource.ResourceHandle
	srcTiDB         *apiresource.ResourceHandle
	dstTiDB         *apiresource.ResourceHandle
	syncDiffConfigs *sdiconfig.Config
	sourceSnapshot  *string // optional, specify it or leave it as ""
	targetSnapshot  *string // optional, specify it or leave it as ""
	timeout         time.Duration
}

func (s syncDiffInspectorCheck) String() string {
	return fmt.Sprintf("run sync diff inspect to check consistency")
}

func (s syncDiffInspectorCheck) Execute(ctx context.Context) error {
	syncDiffConfigFile := "/sync_diff.toml"
	// update sync-diff config with actual upstream and downstream tidb
	src := utils.Try(core.ResolveResourceHandle(s.srcTiDB)).(resource.TiDBCluster)
	srcUri := utils.Try((src).ServiceURL(resource.DBAddr)).(*url.URL)
	s.syncDiffConfigs.DataSources[s.syncDiffConfigs.Task.Source[0]].Host = srcUri.Hostname()
	s.syncDiffConfigs.DataSources[s.syncDiffConfigs.Task.Source[0]].Port = utils.Try(strconv.Atoi(srcUri.Port())).(int)

	dst := utils.Try(core.ResolveResourceHandle(s.dstTiDB)).(resource.TiDBCluster)
	dstUri := utils.Try((dst).ServiceURL(resource.DBAddr)).(*url.URL)
	s.syncDiffConfigs.DataSources[s.syncDiffConfigs.Task.Target].Host = dstUri.Hostname()
	s.syncDiffConfigs.DataSources[s.syncDiffConfigs.Task.Target].Port = utils.Try(strconv.Atoi(dstUri.Port())).(int)

	if *s.sourceSnapshot != "" {
		sources := s.syncDiffConfigs.Task.Source
		for _, sourceName := range sources {
			log.Info(sourceName)
			source := s.syncDiffConfigs.DataSources[sourceName]
			var _ = reflections.SetField(source, "Snapshot", *s.sourceSnapshot)
			if snapshot, err := reflections.GetField(source, "Snapshot"); err != nil {
				return errors.Trace(err)
			} else if snapshot != *s.sourceSnapshot {
				return errors.New("set snapshot failed")
			}
		}
	}

	s.syncDiffConfigs.Task.OutputDir = s.syncDiffConfigs.Task.OutputDir + "_" + time.Now().Format("20060102_150405")
	diffRes := utils.Try(core.ResolveResourceHandle(s.syncDiff)).(resource.WorkloadNode)
	var b bytes.Buffer
	if err := toml.NewEncoder(&b).Encode(s.syncDiffConfigs); err != nil {
		return err
	}
	writeOpts := resource.WorkloadNodeWriteFileOptions{
		Content:  strings.NewReader(b.String()),
		FilePath: syncDiffConfigFile,
	}
	utils.Try(diffRes.WriteFile(writeOpts))

	// run sync-diff check
	res, err := ExecSyncDiffCmd(diffRes, "/sync_diff_inspector -C "+syncDiffConfigFile, s.timeout)
	if err != nil {
		return errors.Trace(err)
	}
	if res.CheckErr() {
		return errors.New("sync diff failed")
	}
	log.Info("SyncDiffInspectorCheck ok!")
	return nil
}

func (b *Builder) SyncDiffInspectorCheck(syncDiff, srcTiDB, dstTiDB *apiresource.ResourceHandle, syncDiffConfigs *sdiconfig.Config, sourceSnapshot, targetSnapshot *string, timeout time.Duration) *Builder {
	b.steps = append(b.steps, &syncDiffInspectorCheck{syncDiff, srcTiDB, dstTiDB, syncDiffConfigs, sourceSnapshot, targetSnapshot, timeout})
	return b
}

func NewSyncDiffInspectorCheck(syncDiff, srcTiDB, dstTiDB *apiresource.ResourceHandle, syncDiffConfigs *sdiconfig.Config, sourceSnapshot, targetSnapshot *string, timeout time.Duration) Step {
	return &syncDiffInspectorCheck{syncDiff, srcTiDB, dstTiDB, syncDiffConfigs, sourceSnapshot, targetSnapshot, timeout}
}