package main

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

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

type operation string

// operation
const (
	KILL operation = "KILL"
	DOWN operation = "DOWN"
)

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

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

// nemesis does dm nemesis
type nemesis struct {
	sync.RWMutex
	caseHost string
	master   string
	workers  []string
	dmctl    *Dmctl
	configs  map[int]string // mapping from caseID to case task config filename

	dmlMu   sync.Mutex // used for controlling dml round
	dmlCh   map[int]chan bool
	dmlStat map[int]int32

	checkMu   sync.Mutex // used for checking round
	checkCh   map[int]chan bool
	checkStat map[int]int32

	emptyMu sync.Mutex // used for idle round
	emptyCh map[int]chan bool

	mssG   *masterSlaveSwitchGroup
	queryG *queryGroup

	errCh chan error
}

// queryGroup is used for query information from DMController to DMCase
type queryGroup struct {
	request struct {
		sync.Mutex
		ch map[int]chan *Message
	}

	response struct {
		sync.Mutex
		ch map[int]chan *MessageResponse
	}
}

// masterSlaveSwitchGroup is used for passing information between DMController
// and DMCase for master slave switch
type masterSlaveSwitchGroup struct {
	sync.Mutex

	precheck struct {
		sync.Mutex
		ch map[int]chan int
	}

	ready struct {
		sync.Mutex
		ch map[int]chan bool
	}

	switch1 struct {
		sync.Mutex
		ch map[int]chan *mssOptions
	}

	switch2 struct {
		sync.Mutex
		ch map[int]chan bool
	}

	done struct {
		sync.Mutex
		ch map[int]chan bool
	}
}

func newNemesis(caseHost, master string, workerIPs []string, dmctl *Dmctl) *nemesis {
	nemesis := &nemesis{
		dmctl:     dmctl,
		caseHost:  caseHost,
		master:    master,
		workers:   workerIPs,
		configs:   make(map[int]string),
		dmlCh:     make(map[int]chan bool),
		checkCh:   make(map[int]chan bool),
		emptyCh:   make(map[int]chan bool),
		dmlStat:   make(map[int]int32),
		checkStat: make(map[int]int32),
		mssG:      &masterSlaveSwitchGroup{},
		queryG:    &queryGroup{},
		errCh:     make(chan error),
	}
	nemesis.mssG.precheck.ch = make(map[int]chan int)
	nemesis.mssG.ready.ch = make(map[int]chan bool)
	nemesis.mssG.switch1.ch = make(map[int]chan *mssOptions)
	nemesis.mssG.switch2.ch = make(map[int]chan bool)
	nemesis.mssG.done.ch = make(map[int]chan bool)

	nemesis.queryG.request.ch = make(map[int]chan *Message)
	nemesis.queryG.response.ch = make(map[int]chan *MessageResponse)

	return nemesis
}

func (n *nemesis) stopWorker(ip string, op operation) error {
	n.Lock()
	defer n.Unlock()

	client.Infof("stop worker %s", ip)
	switch op {
	case KILL:
		return errors.Trace(n.stopProcess(ip, "dm-worker"))
	case DOWN:
		return errors.Trace(n.downProcess(ip, "dm-worker"))
	default:
		return errors.NotSupportedf("stop worker %s operation %s", ip, op)
	}
}

func (n *nemesis) stopMaster(ip string, op operation) error {
	n.Lock()
	defer n.Unlock()

	client.Infof("stop master %s", ip)
	switch op {
	case KILL:
		return errors.Trace(n.stopProcess(ip, "dm-master"))
	case DOWN:
		return errors.Trace(n.downProcess(ip, "dm-master"))
	default:
		return errors.NotSupportedf("stop master %s operation %s", ip, op)
	}
}

func (n *nemesis) startWorkerOnly(ip string) error {
	client.Infof("start worker %s", ip)

	err := n.startProcess(ip, "dm-worker")
	if err != nil {
		return errors.Annotatef(err, "start worker %s", ip)
	}

	return nil
}

func (n *nemesis) startMaster(ip string) error {
	client.Infof("start master %s", ip)
	return errors.Trace(n.startProcess(ip, "dm-master"))
}

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

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

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

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
				}

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

			if err := checkProcessState(cli, processName, daemon.ProcStatRunning); err != nil {
				client.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 {
				client.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.Poll(10*time.Second, 1*time.Minute, func() (bool, error) {
			if err := cli.ShutdownProcess(processName); err != nil {
				client.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.Poll(10*time.Second, 2*time.Minute, func() (bool, error) {
		tmpState, err := cli.ProcessState(processName)
		if err != nil {
			client.Errorf("get %s state failed: %v, retry", processName, err)
			return false, nil
		}

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

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