package main

import (
	"fmt"
	"math/rand"
	"strings"
	"sync"
	"time"

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

const (
	//AgentPort is agent port
	AgentPort = 32337
)

func init() {
	rand.Seed(time.Now().UnixNano())
}

type nemesis struct {
	cli                      *client.Client
	alivePumps               []string
	stoppedPumps             []string
	drainer                  string
	disturb_pump_duration    time.Duration
	kill_pump                bool // weather kill -9
	disturb_drainer_duration time.Duration

	isPause bool
	pauseMu sync.Mutex
}

func (n *nemesis) do() error {
	var pumpC <-chan time.Time
	var drainerC <-chan time.Time

	if n.disturb_pump_duration > 0 {
		pumpTicker := time.NewTicker(disturb_pump_duration)
		defer pumpTicker.Stop()
		pumpC = pumpTicker.C
	}

	if n.disturb_drainer_duration > 0 {
		drainerTicker := time.NewTicker(disturb_drainer_duration)
		defer drainerTicker.Stop()
		drainerC = drainerTicker.C
	}

	for {
		select {
		case <-pumpC:
			n.pauseMu.Lock()
			if n.isPause {
				n.pauseMu.Unlock()
				continue
			}

			log.Debug("pump tick..")
			log.Debug("start rand do pump")
			err := n.randDoPump()
			if err != nil {
				n.pauseMu.Unlock()
				return errors.Trace(err)
			}

			n.pauseMu.Unlock()
		case <-drainerC:
			n.pauseMu.Lock()
			if n.isPause {
				n.pauseMu.Unlock()
				continue
			}
			log.Debug("drainer tick..")

			// just restart drainer
			err := n.downDrainer()
			if err != nil {
				n.pauseMu.Unlock()
				return errors.Trace(err)
			}

			err = n.startDrainer()
			if err != nil {
				n.pauseMu.Unlock()
				return errors.Trace(err)
			}

			n.pauseMu.Unlock()
		}
	}
}

func (n *nemesis) pause() {
	n.pauseMu.Lock()
	n.isPause = true
	n.pauseMu.Unlock()
}

func (n *nemesis) resume() {
	n.pauseMu.Lock()
	n.isPause = false
	n.pauseMu.Unlock()
}

func (n *nemesis) randDoPump() error {
	index := rand.Intn(3)
	switch index {
	case 0:
		log.Debug("stop a pump")
		return errors.Trace(n.stopAPump(n.kill_pump))
	case 1:
		log.Debug("start a pump")
		return errors.Trace(n.startAPump())
	case 2:
		log.Debug("create a pump")
		return errors.Trace(n.newPump())
	}
	return nil
}

func (n *nemesis) newPump() error {
	index := len(n.alivePumps) + len(n.stoppedPumps)
	if index > 5 {
		log.Debug("to many pumps, not create")
		return nil
	}
	pd := n.cli.PdService()
	pump, err := startAPump(n.cli, binlogVersionTag, pd, index)
	if err != nil {
		return errors.Trace(err)
	}

	n.alivePumps = append(n.alivePumps, pump)
	return nil
}

func (n *nemesis) stopAPump(kill bool) error {
	if len(n.alivePumps) <= 1 {
		log.Debugf("not stop pump to keep at least one pump, the pumps size is %d", len(n.alivePumps))
		return nil
	}

	index := rand.Intn(len(n.alivePumps))
	pumpIP := n.alivePumps[index]
	log.Debugf("stop pump %s", pumpIP)
	return errors.Trace(n.stopPump(pumpIP, kill))
}

func (n *nemesis) startPumps() error {
	for _, ip := range n.stoppedPumps {
		err := n.startPump(ip)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func (n *nemesis) startAPump() error {
	if len(n.stoppedPumps) < 1 {
		log.Debugf("not start pump, the size if just %d", len(n.stoppedPumps))
		return nil
	}
	index := rand.Intn(len(n.stoppedPumps))
	pumpIP := n.stoppedPumps[index]
	log.Debugf("start pump %s", pumpIP)
	return errors.Trace(n.startPump(pumpIP))
}

func (n *nemesis) stopPump(ip string, kill bool) error {
	log.Debugf("nemesis :%+v", n)
	n.stoppedPumps = append(n.stoppedPumps, ip)
	n.alivePumps = removeString(n.alivePumps, ip)
	if kill {
		return errors.Trace(n.stopProcess(ip, "pump"))
	} else {
		return errors.Trace(n.downProcess(ip, "pump"))
	}
}

func (n *nemesis) startPump(ip string) error {
	log.Debugf("nemesis :%+v", n)
	n.stoppedPumps = removeString(n.stoppedPumps, ip)
	n.alivePumps = append(n.alivePumps, ip)
	return errors.Trace(n.startProcess(ip, "pump"))
}

func (n *nemesis) downDrainer() error {
	log.Debugf("down drainer nemesis :%+v", n)
	return errors.Trace(n.downProcess(n.drainer, "drainer"))
}

func (n *nemesis) stopDrainer() error {
	log.Debugf("stop drainer nemesis :%+v", n)
	return errors.Trace(n.stopProcess(n.drainer, "drainer"))
}

func (n *nemesis) startDrainer() error {
	log.Debugf("start drainer nemesis :%+v", n)
	return errors.Trace(n.startProcess(n.drainer, "drainer"))
}

func (n *nemesis) startProcess(ip, process string) error {
	clients := genClients([]string{ip})
	return errors.Trace(agentDoRestart(clients, process))
}

func (n *nemesis) execCommand(ip, command string) error {
	clients := genClients([]string{ip})
	cmd := &core.ShellCommand{Command: command}
	return errors.Trace(agentExec(clients, cmd))
}

// kill -9
func (n *nemesis) stopProcess(ip, process string) error {
	clients := genClients([]string{ip})
	return errors.Trace(agentDoKill(clients, process))
}

// kill
func (n *nemesis) downProcess(ip, process string) error {
	clients := genClients([]string{ip})
	return errors.Trace(agentDoDown(clients, process))
}

func removeString(array []string, s string) []string {
	var a []string
	for _, str := range array {
		if str == s {
			continue
		}
		a = append(a, str)
	}
	return a
}

func genClients(ips []string) []*agClient.Client {
	var clients []*agClient.Client
	for _, ip := range ips {
		clients = append(clients, agClient.NewClient(fmt.Sprintf("%s:%d", ip, AgentPort)))
	}
	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.PollImmediate(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 agentDoDown(clients []*agClient.Client, processName string) error {
	for _, cli := range clients {
		err := wait.PollImmediate(10*time.Second, 1*time.Minute, func() (bool, error) {
			if err := cli.ShutdownProcess(processName); err != nil {
				log.Errorf("shutdown %s failed: %v, retry", processName, err)
				return false, nil
			}

			if err := checkProcessState(cli, processName, daemon.ProcStatStopped); 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.PollImmediate(3*time.Second, 2*time.Minute, 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 agentExec(clients []*agClient.Client, command *core.ShellCommand) error {
	for _, cli := range clients {
		data, err := cli.ExecCommand(command)
		if err != nil {
			log.Errorf("exec %+v failed: %v, retry", command, err)
			return errors.Trace(err)
		}

		log.Debug("exec shell return data is %s", string(data))
	}
	return nil
}