package steps

import (
	"context"
	"fmt"
	"net/url"
	"regexp"
	"strings"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap/test-infra/caselib/pkg/host"
	dmmodel "github.com/pingcap/test-infra/caselib/pkg/model/dm"
	"github.com/pingcap/test-infra/caselib/pkg/model/workload"
	"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"
)

type initDMHost struct {
	dm         *apiresource.ResourceHandle
	downStream *apiresource.ResourceHandle
	upStreams  []*apiresource.ResourceHandle
	syncDiff   *apiresource.ResourceHandle
	dmHost     *host.DMHost
}

func (i *initDMHost) Execute(ctx context.Context) error {
	i.dmHost.SetDM(i.dm)
	i.dmHost.SetUpStreams(i.upStreams)
	i.dmHost.SetDownStream(i.downStream)
	i.dmHost.SetSyncDiff(i.syncDiff)
	i.dmHost.Init()
	return nil
}

func (i *initDMHost) String() string {
	return "Init dm host"
}

func (b *Builder) InitDMHost(dm, downStream, syncDiff *apiresource.ResourceHandle, upStreams []*apiresource.ResourceHandle, dmHost *host.DMHost) *Builder {
	b.steps = append(b.steps, &initDMHost{
		dm:         dm,
		downStream: downStream,
		upStreams:  upStreams,
		syncDiff:   syncDiff,
		dmHost:     dmHost,
	})
	return b
}

type adjustConfig struct {
	dm       *apiresource.ResourceHandle
	syncDiff *apiresource.ResourceHandle
	dmHost   *host.DMHost
	path     *dmmodel.Path
}

func (a *adjustConfig) Execute(ctx context.Context) error {
	utils.Try(a.dmHost.AdjustSourceCfg(a.path.SourcePaths))
	utils.Try(a.dmHost.AdjustTaskCfg(a.path.TaskPaths))
	utils.Try(a.dmHost.AdjustDiffCfg(a.path.DiffPaths))
	return nil
}

func (a *adjustConfig) String() string {
	return fmt.Sprintf("adjust config by these file: %+v", a.path)
}

func (b *Builder) AdjustConfig(dm, syncDiff *apiresource.ResourceHandle, dmHost *host.DMHost, path *dmmodel.Path) *Builder {
	b.steps = append(b.steps, &adjustConfig{
		dm:       dm,
		syncDiff: syncDiff,
		dmHost:   dmHost,
		path:     path,
	})
	return b
}

type checkSyncDiff struct {
	dmHost   *host.DMHost
	diffPath string
}

func (c *checkSyncDiff) Execute(ctx context.Context) error {
	return c.dmHost.CheckSyncDiff(c.diffPath, 3)
}

func (c *checkSyncDiff) String() string {
	return "start sync-diff by diff file: " + c.diffPath
}

func (b *Builder) DmCheckSyncDiff(dm, syncDiff *apiresource.ResourceHandle, dmHost *host.DMHost, diffPath string) *Builder {
	b.steps = append(b.steps, &checkSyncDiff{
		dmHost:   dmHost,
		diffPath: diffPath,
	})
	return b
}

// matchQueryStatusStep this step retry 10 times for every 6s
type matchQueryStatusStep struct {
	dmHost   *host.DMHost
	taskName string
	re       *regexp.Regexp
	count    int
}

func NewMatchQueryStatusStep(dmHost *host.DMHost, taskName, regexStr string, count int) *matchQueryStatusStep {
	re, err := regexp.Compile(regexStr)
	if err != nil {
		panic(err)
	}
	return &matchQueryStatusStep{
		dmHost:   dmHost,
		taskName: taskName,
		re:       re,
		count:    count,
	}
}

func (c *matchQueryStatusStep) String() string {
	return "check query-status"
}

func (c *matchQueryStatusStep) Execute(ctx context.Context) error {
	for i := 0; i < 10; i++ {
		result, _, err := c.dmHost.QueryStatus(c.taskName)
		if err != nil {
			continue
		}
		matches := c.re.FindAllString(result, -1)
		if len(matches) != c.count {
			log.Info(fmt.Sprintf("query status failed, expect %d, got %d", c.count, len(matches)))
			time.Sleep(6 * time.Second)
			continue
		}
		return nil
	}
	return fmt.Errorf("query status failed")
}

func (b *Builder) DmMatchQueryStatusStep(dmHost *host.DMHost, taskName, regexStr string, count int) *Builder {
	b.steps = append(b.steps, NewMatchQueryStatusStep(dmHost, taskName, regexStr, count))
	return b
}

type dmctlCommandStep struct {
	dmHost *host.DMHost
	cmd    string
}

func NewDmctlCommandStep(dmHost *host.DMHost, cmd string) *dmctlCommandStep {
	return &dmctlCommandStep{
		dmHost: dmHost,
		cmd:    cmd,
	}
}

func (d *dmctlCommandStep) String() string {
	return "running dmctl command: " + d.cmd
}

var dmctlCommandStepRegex = regexp.MustCompile(`"result": false`)

func (d *dmctlCommandStep) Execute(ctx context.Context) error {
	res, err := d.dmHost.ExecDmctlCmd(d.cmd, host.TIMEOUT)
	if err != nil {
		return err
	}
	if len(dmctlCommandStepRegex.FindAllString(res.Stdout, -1)) != 0 {
		return fmt.Errorf("dmctl command failed")
	}
	return nil
}

func (b *Builder) DmDmCtlCmd(dmHost *host.DMHost, cmd string) *Builder {
	b.steps = append(b.steps, NewDmctlCommandStep(dmHost, cmd))
	return b
}

type oscType int

const (
	PtOsc oscType = iota
	GhostOsc
	OscTimeout = 10 * time.Minute
)

type onlineSchemaChangeStep struct {
	upStream  *apiresource.ResourceHandle
	oscHandle *apiresource.ResourceHandle
	osc       oscType
	alter     string
	database  string
	table     string
}

func NewOnlineSchemaChangeStep(upStream, oscHandle *apiresource.ResourceHandle, osc oscType, alter, database, table string) *onlineSchemaChangeStep {
	return &onlineSchemaChangeStep{
		upStream:  upStream,
		oscHandle: oscHandle,
		osc:       osc,
		alter:     alter,
		database:  database,
		table:     table,
	}
}

func (o *onlineSchemaChangeStep) String() string {
	return fmt.Sprintf("online schema change %v, alter=%s", o.osc, o.alter)
}

func (o *onlineSchemaChangeStep) Execute(ctx context.Context) error {
	up := utils.Try(core.ResolveResourceHandle(o.upStream)).(resource.MySQL)
	dbURL := utils.Try(up.ServiceURL(resource.DBAddr)).(*url.URL)
	wl := utils.Try(core.ResolveResourceHandle(o.oscHandle)).(resource.WorkloadNode)
	workloadHost := host.NewWorkloadHost(wl)

	dbHost := strings.Split(dbURL.Host, ":")[0]

	var argv []string
	// it's normal to report error: xxx_ghc' doesn't exist
	// https://github.com/github/gh-ost/issues/657
	if o.osc == GhostOsc {
		argv = []string{
			"gh-ost",
			"--user=root",
			"--host=" + dbHost,
			"--allow-on-master",
			"--database=" + o.database,
			"--table=" + o.table,
			"--ok-to-drop-table",
			"--alter=" + fmt.Sprintf(`"%s"`, o.alter),
			"--default-retries=120",
			"--serve-socket-file=/tmp/" + dbHost,
			"--execute",
		}
	} else {
		// it's normal to report: Cannot connect to h=xxx,u=root
		// seems pt-osc is connecting other slaves, but skip-check-slave-lag is not working
		argv = []string{
			"pt-online-schema-change",
			"--execute",
			"--alter=" + fmt.Sprintf(`"%s"`, o.alter),
			fmt.Sprintf("h=%s,u=root,D=%s,t=%s", dbHost, o.database, o.table),
		}
	}

	if err := workloadHost.Run(workload.NewCMD(strings.Join(argv, " "), OscTimeout)); err != nil {
		return err
	}
	return nil
}

func (b *Builder) DmOnlineSchemaChangeStep(upStream, oscHandle *apiresource.ResourceHandle, osc oscType, alter, database, table string) *Builder {
	b.steps = append(b.steps, NewOnlineSchemaChangeStep(upStream, oscHandle, osc, alter, database, table))
	return b
}

type disableGtid struct {
	upStream *apiresource.ResourceHandle
	dmHost   *host.DMHost
}

func (c *disableGtid) Execute(ctx context.Context) error {
	mysql := utils.Try(core.ResolveResourceHandle(c.upStream)).(resource.MySQLCompatible)
	mysqlUrl := utils.Try(mysql.ServiceURL(resource.DBAddr)).(*url.URL)
	gtidStatus, err := c.dmHost.GetGtidStatus(c.upStream)
	if err != nil {
		return err
	}
	if gtidStatus == "OFF" {
		return nil
	}
	return c.dmHost.DisableGtid(mysqlUrl)
}

func (c *disableGtid) String() string {
	return "disable gtid for mysql. "
}

func (b *Builder) DmDisableGtid(upStream *apiresource.ResourceHandle, dmHost *host.DMHost) *Builder {
	b.steps = append(b.steps, &disableGtid{
		dmHost:   dmHost,
		upStream: upStream,
	})
	return b
}

type upgradeDmCluster struct {
	dm      *apiresource.ResourceHandle
	dmHost  *host.DMHost
	version *string
	timeout time.Duration
}

func NewUpgradeDmCluster(dmHost *host.DMHost, dm *apiresource.ResourceHandle, version *string, timeout time.Duration) Step {
	return &upgradeDmCluster{
		dmHost:  dmHost,
		dm:      dm,
		version: version,
		timeout: timeout,
	}
}

func (s *upgradeDmCluster) Execute(ctx context.Context) error {
	dmRes := utils.Try(core.ResolveResourceHandle(s.dm)).(resource.DMCluster)

	// Upgrade tidb
	start := time.Now()
	if err := dmRes.Upgrade(resource.DMClusterUpgradeOptions{
		Version: *s.version,
		Timeout: s.timeout,
	}); err != nil {
		log.Error("upgrade Dmcluster failed", zap.Error(err), zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("upgrade tidb cluster success", zap.Any("option", s), zap.Duration("elapsed", time.Since(start)))

	return nil
}

func (s *upgradeDmCluster) String() string {
	versionStr := "<<???>>"
	if s.version != nil {
		versionStr = *s.version
	}
	return fmt.Sprintf("Upgrade DM to version=%s", versionStr)
}

func (s *Builder) DmUpgradeDmCluster(dmHost *host.DMHost, dm *apiresource.ResourceHandle, version *string, timeout time.Duration) *Builder {
	s.steps = append(s.steps, NewUpgradeDmCluster(dmHost, dm, version, timeout))
	return s
}

type count1GBinlogConsumeDuration struct {
	dmHost     *host.DMHost
	workerName *string
	duration   *string
}

func (c *count1GBinlogConsumeDuration) Execute(ctx context.Context) error {
	durationTime, err := c.dmHost.Get1GBDMBinlogConsumeDuration(*c.workerName)
	if err != nil {
		return err
	}
	*c.duration = durationTime.Stdout
	fmt.Printf("IS worker: %s, duration: %s", *c.workerName, *c.duration)
	return nil
}

func (c *count1GBinlogConsumeDuration) String() string {
	return fmt.Sprintf("worker: %s, duration: %s", *c.workerName, *c.duration)
}

func NewCount1GBinlogConsumeDuration(dmHost *host.DMHost, workerName *string, duration *string) *count1GBinlogConsumeDuration {
	return &count1GBinlogConsumeDuration{
		dmHost,
		workerName,
		duration,
	}
}

func (b *Builder) Count1GBinlogConsumeDuration(dmHost *host.DMHost, workerName *string, duration *string) *Builder {
	b.steps = append(b.steps, NewCount1GBinlogConsumeDuration(dmHost, workerName, duration))
	return b
}

type getDMWorkerVersion struct {
	dmHost     *host.DMHost
	workerName *string
	version    *string
	hashcode   *string
}

func (c *getDMWorkerVersion) Execute(ctx context.Context) error {
	res, err := c.dmHost.DMWorkerVersion(*c.workerName)
	if err != nil {
		return err
	}
	// *c.version
	workerVersion := res.Stdout
	workVersion := workerVersion[strings.Index(workerVersion, ":")+2 : strings.Index(workerVersion, "\n")]
	hashCodeVersion := workerVersion[strings.Index(workerVersion, "Hash:")+6 : strings.Index(workerVersion, "Git Branch")-1]
	*c.version = workVersion
	*c.hashcode = hashCodeVersion
	fmt.Printf("IS worker: %s, verioninfo: %s, hashcode: %s", *c.workerName, *c.version, *c.hashcode)
	return nil
}

func (c *getDMWorkerVersion) String() string {
	return fmt.Sprintf("worker: %s, duration: %s", *c.workerName, *c.version)
}

func NewGetDMWorkerVersion(dmHost *host.DMHost, workerName, version, hashcode *string) *getDMWorkerVersion {
	return &getDMWorkerVersion{
		dmHost,
		workerName,
		version,
		hashcode,
	}
}

func (b *Builder) GetDMWorkerVersion(dmHost *host.DMHost, workerName, version, hashcode *string) *Builder {
	b.steps = append(b.steps, NewGetDMWorkerVersion(dmHost, workerName, version, hashcode))
	return b
}