package client

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	agClient "github.com/pingcap/schrodinger/agent/client"
	"github.com/pingcap/schrodinger/agent/daemon"
	"github.com/pingcap/schrodinger/pkg/types"
	"github.com/pingcap/schrodinger/util"
	"k8s.io/apimachinery/pkg/util/wait"
)

const (
	// RetryCount is the maximum retry count.
	RetryCount = 5

	// RetryInterval is the interval that each retry must wait.
	RetryInterval = 1 * time.Second

	clusterAgentPort = 8350

	daemonPrefix = "api/v1"
)

func postAndExtract(url string, postData []byte) ([]byte, error) {
	var data []byte
	d, err := util.DoPost(http.DefaultClient, url, postData)
	if err != nil {
		return data, errors.Annotatef(err, "url - %s", url)
	}
	data, err = util.ExtractResponse(d)
	return data, errors.Trace(err)
}

func getAndExtract(url string) ([]byte, error) {
	var data []byte
	d, err := util.DoGet(http.DefaultClient, url)
	if err != nil {
		return data, errors.Annotatef(err, "url - %s", url)
	}
	data, err = util.ExtractResponse(d)
	return data, errors.Trace(err)
}

func deleteAndExtract(url string) ([]byte, error) {
	var data []byte
	d, err := util.DoDelete(http.DefaultClient, url)

	if err != nil {
		return data, errors.Annotatef(err, "url - %s", url)
	}

	data, err = util.ExtractResponse(d)
	return data, errors.Trace(err)
}

func putAndExtract(url string, putData []byte) ([]byte, error) {
	var data []byte
	d, err := util.DoPut(http.DefaultClient, url, putData)
	if err != nil {
		return data, errors.Annotatef(err, "url - %s", url)
	}
	data, err = util.ExtractResponse(d)
	return data, errors.Trace(err)
}

func (c *Client) genPodInfo(pods []types.PodStatus, typeName string) []*PodInfo {
	var podInfos []*PodInfo

	for _, p := range pods {
		info := &PodInfo{
			Name:        p.Name,
			IP:          p.PodIP,
			NodeIP:      p.NodeIP,
			ContainerID: p.ContainerID,
			AgentPort:   c.conf.Manager.AgentPort,
		}

		switch typeName {
		case TiDB:
			info.ServicePort = c.conf.Manager.TidbServicePort
		case PD:
			info.ServicePort = c.conf.Manager.PdServicePort
			info.DataDir = c.conf.Manager.DataDir
		case TiKV:
			info.ServicePort = c.conf.Manager.TikvServicePort
			info.DataDir = c.conf.Manager.DataDir
		default:
			Errorf("type %s is not valid", typeName)
			continue
		}

		podInfos = append(podInfos, info)
	}

	return podInfos
}

func (c *Client) waitingCatStageIsRunning(interval, timeout time.Duration, expectStage types.CatStage) error {
	err := wait.Poll(interval, timeout, func() (bool, error) {
		data, err := getAndExtract(fmt.Sprintf("%s/cats/%d/status", c.managerAddr, c.conf.Cat.ID))
		if err != nil {
			if strings.Contains(err.Error(), "not found") {
				return true, errors.Trace(err)
			}

			Errorf("get cat %d stage failed: %v, retry", c.conf.Cat.ID, err)
			return false, nil
		}
		status := &types.CatStatus{}

		if err = json.Unmarshal(data, status); err != nil {
			Errorf("get cat %d status %v failed: %v", c.conf.Cat.ID, status, err)
			return true, errors.Errorf("get cat %d status %v failed: %v", c.conf.Cat.ID, status, err)
		}

		switch status.Stage {
		case expectStage:
			if status.State != types.CatStateNormal {
				Errorf("cat %d status %s", c.conf.Cat.ID, status.State)
				return true, errors.Errorf("cat %d status %s", c.conf.Cat.ID, status.State)
			}
			Debugf("get cat %d stage is %s, retry", c.conf.Cat.ID, expectStage)
			return false, nil
		case types.CatStageRunning:
			return true, nil
		default:
			return true, errors.Errorf("get cat %d stage is %s, expect stage is %s or %s", c.conf.Cat.ID, status.Stage, expectStage, types.CatStageRunning)
		}
	})

	return errors.Trace(err)
}

func retryOnError(ctx context.Context, fn func() error) error {
	var err error
	for i := 0; i < RetryCount; i++ {
		select {
		case <-ctx.Done():
			return nil
		default:
		}
		err = fn()
		if err == nil {
			break
		}

		Error(err)
		sleep(ctx, RetryInterval)
	}

	return errors.Trace(err)
}

func sleep(ctx context.Context, sleepTime time.Duration) {
	ticker := time.NewTicker(sleepTime)
	defer ticker.Stop()

	select {
	case <-ctx.Done():
		return
	case <-ticker.C:
		return
	}
}

func checkNodeState(ctx context.Context, state daemon.ProcessState, agentCli *agClient.Client, processName string) bool {
	ctx, cancel := context.WithTimeout(ctx, 5*time.Minute)
	defer cancel()

	for {
		select {
		case <-ctx.Done():
			return false
		default:
		}

		tmpState, err := agentCli.ProcessState(processName)
		if err != nil {
			Errorf("get %s state failed: %v", processName, err)
		} else {
			Infof("get %s state: %s, expect state: %s", processName, tmpState.String(), state.String())
			if tmpState == state {
				return true
			}
		}

		sleep(ctx, 10*time.Second)
	}
}

func daemonAction(ctx context.Context, url string, data []byte) error {
	var err error
	for i := 0; i < RetryCount; i++ {
		select {
		case <-ctx.Done():
			return nil
		default:
		}
		var resp *Response

		Infof("post url: %s, data: %s", url, data)
		resp, err = doPost(url, data)
		if err != nil {
			err = errors.Errorf("do post failed: %v", err)
			Error(err)
			sleep(ctx, RetryInterval)
			continue
		}

		if resp.Code == http.StatusOK {
			return nil
		}

		if resp.Code == http.StatusNotFound {
			return errors.Errorf("resources do not exist: %v", err)
		}

		err = errors.Errorf("do post failed: code %d, message: %s", resp.Code, resp.Message)
		Error(err)
		sleep(ctx, RetryInterval)
	}
	return errors.Trace(err)
}

// Response represents message that returns to client
type Response struct {
	Code    int         `json:"code"`
	Message string      `json:"message"`
	Data    interface{} `json:"data"`
}

func doPost(url string, data []byte) (*Response, error) {
	resp, err := util.DoPost(http.DefaultClient, url, data)
	if err != nil {
		return nil, errors.Trace(err)
	}

	respData := &Response{}
	if err := json.Unmarshal(resp, respData); err != nil {
		return nil, errors.Trace(err)
	}

	return respData, nil
}

func genFakeCatInfo(tidbServiceStr, pdServiceStr, tidbStr, pdStr, tikvStr string) (*CatInfo, error) {
	var (
		tidbs []*PodInfo
		pds   []*PodInfo
		tikvs []*PodInfo
		err   error
	)

	if err = checkTiDBService(tidbServiceStr); err != nil {
		return nil, errors.Trace(err)
	}

	tidbs, err = genFakePodInfo(tidbStr)
	if err != nil {
		return nil, errors.Trace(err)
	}

	pds, err = genFakePodInfo(pdStr)
	if err != nil {
		return nil, errors.Trace(err)
	}

	tikvs, err = genFakePodInfo(tikvStr)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return &CatInfo{
		LBService: tidbServiceStr,
		PdService: pdServiceStr,
		TiDBs:     tidbs,
		PDs:       pds,
		TiKVs:     tikvs,
	}, nil
}

func genFakePodInfo(str string) ([]*PodInfo, error) {
	var podInfos []*PodInfo

	ips, ports, err := genIPsAndPorts(str)
	if err != nil {
		return podInfos, errors.Trace(err)
	}

	if len(ips) != len(ports) {
		return podInfos, errors.NotValidf("addr string %s", str)
	}

	for index, ip := range ips {
		info := &PodInfo{
			IP:          ip,
			ServicePort: ports[index],
		}

		podInfos = append(podInfos, info)
	}

	return podInfos, nil
}

func checkTiDBService(tidbStr string) error {
	if util.IsEmpty(tidbStr) {
		return nil
	}

	arr := strings.Split(tidbStr, ":")
	if len(arr) < 2 {
		return errors.NotValidf("tidb service %s", tidbStr)
	}

	_, err := strconv.Atoi(arr[1])
	if err != nil {
		return errors.Trace(err)
	}

	return nil
}

func genIPsAndPorts(str string) ([]string, []int, error) {
	var (
		ips   []string
		ports []int
	)

	if util.IsEmpty(str) {
		return ips, ports, nil
	}

	addrs := strings.Split(str, ",")

	for _, addr := range addrs {
		arr := strings.Split(addr, ":")
		if len(arr) < 2 {
			return ips, ports, errors.NotValidf("addr %s", addr)
		}

		ips = append(ips, arr[0])
		port, err := strconv.Atoi(arr[1])
		if err != nil {
			return ips, ports, errors.Trace(err)
		}
		ports = append(ports, port)
	}

	return ips, ports, nil
}

func genMysqlName(version string) string {
	return strings.Replace(fmt.Sprintf("mysql-%s", version), ".", "-", -1)
}

func genClients(ips []string) []*agClient.Client {
	var clients []*agClient.Client
	for _, ip := range ips {
		clients = append(clients, agClient.NewClient(fmt.Sprintf("%s:%d", ip, clusterAgentPort)))
	}
	return clients
}

func agentDoRestart(clients []*agClient.Client, processName string) error {
	for _, cli := range clients {
		err := wait.Poll(20*time.Second, 5*time.Minute, func() (bool, error) {
			if err := cli.RestartProcess(processName); err != nil {
				if strings.Contains(err.Error(), "start process from state [RUNNING]") {
					return true, nil
				}

				log.Errorf("restart %s failed: %v, retry", processName, err)
				return false, nil
			}

			if err := checkProcessState(cli, processName, daemon.ProcStatRunning); err != nil {
				log.Errorf("check %s expect status %s failed, error: %v, retry", processName, daemon.ProcStatRunning, err)
				return false, nil
			}
			return true, nil
		})

		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func agentDoKill(clients []*agClient.Client, processName string) error {
	for _, cli := range clients {
		err := wait.Poll(10*time.Second, 1*time.Minute, func() (bool, error) {
			if err := cli.StopProcess(processName); err != nil {
				log.Errorf("stop %s failed: %v, retry", processName, err)
				return false, nil
			}

			if err := checkProcessState(cli, processName, daemon.ProcStatKilled); err != nil {
				return true, errors.Trace(err)
			}
			return true, nil
		})

		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func checkProcessState(cli *agClient.Client, processName string, state daemon.ProcessState) error {
	err := wait.Poll(10*time.Second, 30*time.Second, func() (bool, error) {
		tmpState, err := cli.ProcessState(processName)
		if err != nil {
			log.Errorf("get %s state failed: %v, retry", processName, err)
			return false, nil
		}

		if tmpState != state {
			log.Errorf("get %s state: %s, expect state: %s, retry", processName,
				tmpState.String(), state.String())
			return false, nil
		}

		return true, nil
	})
	return errors.Trace(err)
}

func isTiDBOwner(ip string) (bool, error) {
	resp, err := util.DoGet(http.DefaultClient, fmt.Sprintf("http://%s:10080/info", ip))
	if err != nil {
		return false, errors.Trace(err)
	}
	s := make(map[string]interface{})
	err = json.Unmarshal(resp, &s)
	if err != nil {
		return false, errors.Trace(err)
	}
	return s["is_owner"].(bool), nil
}

func getPdLeader(ip string) (string, error) {
	resp, err := util.DoGet(http.DefaultClient, fmt.Sprintf("http://%s:2379/pd/api/v1/leader", ip))
	if err != nil {
		return "", errors.Trace(err)
	}
	s := make(map[string]interface{})
	err = json.Unmarshal(resp, &s)
	if err != nil {
		return "", errors.Trace(err)
	}
	return s["name"].(string), nil
}