package main

import (
	"context"
	"encoding/json"

	"github.com/pingcap/schrodinger-test/dmctl"

	"fmt"
	"regexp"
	"time"

	"github.com/juju/errors"
	"github.com/pingcap/schrodinger-test/dmctl/pb"
	client "github.com/pingcap/schrodinger/pkg/test_client"
	"github.com/pingcap/schrodinger/util"
)

const (
	timeout = 30 * time.Second
)

var (
	promptRE = regexp.MustCompile("»")
)

// Dmctl is dmctl
type Dmctl struct {
	master string
	path   string

	ctl *dmctl.DMMasterCtl
}

// SubTaskError is minimal subtask syn error info to used for sql replace operator
type SubTaskError struct {
	binlogPos string
	errorSQL  string
}

func setUpDmCtl(ctx context.Context, master string) (*Dmctl, error) {
	err := util.InstallArchive(ctx, downloadAddr, "/dmctl")
	if err != nil {
		return nil, errors.Trace(err)
	}

	ctl, err := dmctl.CreateDMMasterCtl(fmt.Sprintf("%s:%d", master, DMMasterPort))
	if err != nil {
		return nil, errors.Trace(err)
	}

	return &Dmctl{
		master: master,
		path:   "/dmctl/bin/dmctl",
		ctl:    ctl,
	}, nil
}

// StartTask starts task
func (dm *Dmctl) StartTask(configFile string) error {
	client.Infof("start task %s", configFile)

	return errors.Trace(dm.ctl.StartTask(context.Background(), configFile, nil))
}

// StartTaskOnWorker starts task on worker
func (dm *Dmctl) StartTaskOnWorker(configFile string, ip string) error {
	worker := fmt.Sprintf("%s:%d", ip, DMWorkerPort)
	client.Infof("start task %s on %s", configFile, worker)
	return errors.Trace(dm.ctl.StartTask(context.Background(), configFile, []string{worker}))
}

func (dm *Dmctl) stopTask(taskName string) error {
	/*cmd := fmt.Sprintf("stop-task %s \n", taskName)
	client.Infof("stop cmd %s", cmd)
	return errors.Trace(dm.execCmd(cmd))*/
	client.Infof("stop task %s", taskName)
	return errors.Trace(dm.ctl.StopTask(context.Background(), taskName, nil))
}

func (dm *Dmctl) existsUnresolvedGroups(taskName string) (bool, error) {
	resp, err := dm.ctl.QueryUnresolvedGroups(context.Background(), taskName)
	if err != nil {
		return false, errors.Trace(err)
	}
	if len(resp) > 0 {
		return true, nil
	}
	return false, nil
}

func (dm *Dmctl) queryStatus() (string, error) {
	responses, err := dm.ctl.QueryStatus(context.Background(), "")
	if err != nil {
		return "", errors.Trace(err)
	}

	rawText, err := json.MarshalIndent(responses, "\t", "\t")
	if err != nil {
		return "", errors.Trace(err)
	}

	return string(rawText), nil
}

// querySyncError returns a map mapping as following
// worker -> task -> SubTaskError pointer
func (dm *Dmctl) querySyncError() (map[string]map[string]*SubTaskError, error) {
	responses, err := dm.ctl.QueryError(context.Background(), "")
	if err != nil {
		return nil, errors.Trace(err)
	}
	result := make(map[string]map[string]*SubTaskError)
	for worker, resp := range responses {
		if !resp.GetResult() {
			return nil, errors.Errorf("fail to query sync error from worker %s: %s", resp.GetWorker(), resp.GetMsg())
		}
		result[worker] = make(map[string]*SubTaskError)
		for _, s := range resp.GetSubTaskError() {
			if s.GetUnit() == pb.UnitType_Sync {
				if sync := s.GetSync(); sync != nil && len(sync.GetErrors()) > 0 {
					syncErr := sync.GetErrors()[0]
					result[worker][s.GetName()] = &SubTaskError{
						// the first failed binlog position is enough
						binlogPos: syncErr.GetFailedBinlogPosition(),
						errorSQL:  syncErr.GetErrorSQL(),
					}
				}
			}
		}
	}
	return result, nil
}

func (dm *Dmctl) queryTasksUnit() (map[string]map[string]pb.UnitType, error) {
	responses, err := dm.ctl.QueryStatus(context.Background(), "")
	if err != nil {
		return nil, errors.Trace(err)
	}
	result := make(map[string]map[string]pb.UnitType)
	for worker, resp := range responses {
		if !resp.GetResult() {
			return nil, errors.Errorf("fail to query task unit %s: %s", resp.GetWorker(), resp.GetMsg())
		}
		if _, ok := result[worker]; !ok {
			result[worker] = make(map[string]pb.UnitType)
		}
		for _, s := range resp.GetSubTaskStatus() {
			result[worker][s.Name] = s.GetUnit()
		}
	}
	return result, nil
}

func (dm *Dmctl) sqlSkipPattern(taskName, worker, sqlPattern string, sharding bool) error {
	err := dm.sqlSkip(taskName, worker, "", sqlPattern, sharding)
	if err != nil {
		return errors.Trace(err)
	}
	client.Infof("worker %s skip sql pattern %s in sharding mode %v", worker, sqlPattern, sharding)
	return nil
}

func (dm *Dmctl) sqlSkipPos(taskName, worker, pos string, sharding bool) error {
	err := dm.sqlSkip(taskName, worker, pos, "", false)
	if err != nil {
		return errors.Trace(err)
	}
	client.Infof("worker %s skip sql at pos %s", worker, pos)
	return nil
}

func (dm *Dmctl) sqlSkip(taskName, worker, pos, sqlPattern string, sharding bool) error {
	_, err := dm.ctl.SkipSQL(context.Background(), taskName, worker, pos, sqlPattern, sharding)
	return errors.Trace(err)
}

func (dm *Dmctl) sqlReplaceWithPos(taskName, worker, pos string, args []string) error {
	// sqlPattern is "", sharding is false
	client.Infof("task %s worker %s sql replace with %v at position %s", taskName, worker, args, pos)
	_, err := dm.ctl.ReplaceSQL(context.Background(), taskName, worker, pos, "", false, args)
	return errors.Trace(err)
}

func (dm *Dmctl) resumeTask(taskName string, workers []string) error {
	if len(workers) == 0 {
		return nil
	}
	_, err := dm.ctl.ResumeTask(context.Background(), taskName, workers)
	if err != nil {
		return errors.Errorf("resume task failed %s", err)
	}
	client.Infof("workers %v task resumed", workers)
	return nil
}

func (dm *Dmctl) updateDmMasterConfig(cfg string) error {
	err := dm.ctl.UpdateMasterConfig(context.Background(), cfg)
	if err != nil {
		return errors.Trace(err)
	}

	return nil
}