package client

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

	"github.com/juju/errors"
	"github.com/pingcap/kvproto/pkg/debugpb"
	agClient "github.com/pingcap/schrodinger/agent/client"
	"github.com/pingcap/schrodinger/agent/daemon"
	"google.golang.org/grpc"
)

// Nemesis is the set of nemesis nodes
type Nemesis struct {
	pdNodes   []*Node
	tidbNodes []*Node
	tikvNodes []*Node
	sync.RWMutex
}

func (n *Nemesis) init(pdNodes, tidbNodes, tikvNodes []*Node) {
	n.Lock()
	n.pdNodes = pdNodes
	n.tidbNodes = tidbNodes
	n.tikvNodes = tikvNodes
	n.Unlock()
}

func (n *Nemesis) selectNode() *Node {
	n.RLock()
	defer n.RUnlock()

	allNodes := make([]*Node, 0, len(n.pdNodes)+len(n.tidbNodes)+len(n.tikvNodes))

	for _, node := range n.pdNodes {
		allNodes = append(allNodes, node)
	}

	for _, node := range n.tidbNodes {
		allNodes = append(allNodes, node)
	}

	for _, node := range n.tikvNodes {
		allNodes = append(allNodes, node)
	}
	randNode := allNodes[rand.Intn(len(allNodes))].clone()

	return randNode
}

func (n *Nemesis) selectNodeByKind(kind string, leaderBiased bool) *Node {
	var node *Node

	switch kind {
	case PD:
		node = n.selectPDNode(leaderBiased)
	case TiKV:
		node = n.selectTiKVNode()
	case TiDB:
		node = n.selectTiDBNode(leaderBiased)
	default:
		return nil
	}
	Infof("select %s node %s(%s)", kind, node.name, node.podInfo.IP)

	return node
}

func (n *Nemesis) selectPDNode(leaderBiased bool) *Node {
	n.RLock()
	defer n.RUnlock()

	randNode := n.pdNodes[rand.Intn(len(n.pdNodes))].clone()

	if leaderBiased {
		var pdLeaderName string

		for _, node := range n.pdNodes {
			name, err := getPdLeader(node.podInfo.IP)
			if err != nil {
				Warnf("query PD leader fails: %s", errors.ErrorStack(err))
			} else {
				pdLeaderName = name
				break
			}
		}

		for _, node := range n.pdNodes {
			if node.podInfo.Name == pdLeaderName {
				return node.clone()
			}
		}
	}

	return randNode
}

func (n *Nemesis) selectTiDBNode(leaderBiased bool) *Node {
	n.RLock()
	defer n.RUnlock()

	randNode := n.tidbNodes[rand.Intn(len(n.tidbNodes))].clone()

	if leaderBiased {
		for _, node := range n.tidbNodes {
			ip := node.podInfo.IP
			isOwner, err := isTiDBOwner(ip)
			if err != nil {
				Warnf("query TiDB owner fails: %s", errors.ErrorStack(err))
				continue
			}
			if isOwner {
				randNode = node.clone()
				break
			}
		}
	}
	return randNode
}

func (n *Nemesis) selectTiKVNode() *Node {
	n.RLock()
	defer n.RUnlock()

	randNode := n.tikvNodes[rand.Intn(len(n.tikvNodes))].clone()

	return randNode
}

// Node is the unit of nemesis execution.
type Node struct {
	name    string
	kind    string
	podInfo *PodInfo

	cli   *agClient.Client
	State daemon.ProcessState

	sync.RWMutex
	ctx context.Context
}

// Name returns name of this Node
func (n *Node) Name() string {
	n.RLock()
	defer n.RUnlock()

	return n.name
}

// Host returns service host of this Node
func (n *Node) Host() string {
	n.RLock()
	defer n.RUnlock()

	return fmt.Sprintf("%s:%d", n.podInfo.IP, n.podInfo.ServicePort)
}

func (n *Node) clone() *Node {
	n.RLock()
	defer n.RUnlock()

	nodeCopy := &Node{
		name:    n.name,
		kind:    n.kind,
		cli:     n.cli,
		ctx:     n.ctx,
		podInfo: n.podInfo.clone(),
	}

	return nodeCopy
}

func (n *Node) start(ctx context.Context) error {
	n.Lock()
	defer n.Unlock()

	Infof("KillNemesis starts service %s %s:%d", n.name, n.podInfo.IP, n.podInfo.ServicePort)
	err := retryOnError(ctx, func() error {
		err := n.cli.RestartProcess(n.kind)
		if err == nil || strings.Contains(err.Error(), "is running") {
			return nil
		}

		return errors.Trace(err)
	})

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

	if !checkNodeState(n.ctx, daemon.ProcStatRunning, n.cli, n.kind) {
		return errors.Errorf("%s(%s) state is not %s", n.kind, n.name, daemon.ProcStatRunning.String())
	}

	return nil
}

func (n *Node) stop(ctx context.Context) error {
	n.Lock()
	defer n.Unlock()

	Infof("KillNemesis kills service %s(%s) %s:%d", n.kind, n.name, n.podInfo.IP, n.podInfo.ServicePort)

	return errors.Trace(n.kill(ctx))
}

func (n *Node) shutDown(ctx context.Context) error {
	err := retryOnError(ctx, func() error {
		return errors.Trace(n.cli.ShutdownProcess(n.kind))
	})

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

	if !checkNodeState(n.ctx, daemon.ProcStatStopped, n.cli, n.kind) {
		return errors.Errorf("%s(%s) state is not %s", n.kind, n.name, daemon.ProcStatStopped.String())
	}

	return nil
}

func (n *Node) kill(ctx context.Context) error {
	err := retryOnError(ctx, func() error {
		return errors.Trace(n.cli.StopProcess(n.kind))
	})

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

	if !checkNodeState(n.ctx, daemon.ProcStatKilled, n.cli, n.kind) {
		return errors.Errorf("%s(%s) state is not %s", n.kind, n.name, daemon.ProcStatKilled.String())
	}

	return nil
}

func (n *Node) pauseProcess(ctx context.Context) error {
	err := retryOnError(ctx, func() error {
		return errors.Trace(n.cli.PauseProcess(n.kind))
	})
	if err != nil {
		return errors.Trace(err)
	}

	if !checkNodeState(n.ctx, daemon.ProcStatPaused, n.cli, n.kind) {
		return errors.Errorf("%s(%s) state is not %s", n.kind, n.name, daemon.ProcStatPaused.String())
	}

	return nil
}

func (n *Node) continueProcess(ctx context.Context) error {
	err := retryOnError(ctx, func() error {
		return errors.Trace(n.cli.ContinueProcess(n.kind))
	})
	if err != nil {
		return errors.Trace(err)
	}

	if !checkNodeState(n.ctx, daemon.ProcStatRunning, n.cli, n.kind) {
		return errors.Errorf("%s(%s) state is not %s", n.kind, n.name, daemon.ProcStatRunning.String())
	}

	return nil
}

func (n *Node) failTiKV(ctx context.Context, fail, actions string) error {
	n.Lock()
	defer n.Unlock()

	addr := fmt.Sprintf("%s:%d", n.podInfo.IP, n.podInfo.ServicePort)
	conn, err := grpc.Dial(addr, grpc.WithInsecure())
	if err != nil {
		return errors.Trace(err)
	}
	defer conn.Close()

	client := debugpb.NewDebugClient(conn)

	// Inject
	injectReq := &debugpb.InjectFailPointRequest{
		Name:    fail,
		Actions: actions,
	}
	Infof("FailNemesis injects service %s(%s):%s, fp: %+v", n.kind, n.name, addr, injectReq)

	_, err = client.InjectFailPoint(ctx, injectReq)
	return errors.Trace(err)
}

func (n *Node) recoverTiKV(ctx context.Context, fail string) error {
	n.Lock()
	defer n.Unlock()

	// Build connections
	addr := fmt.Sprintf("%s:%d", n.podInfo.IP, n.podInfo.ServicePort)
	conn, err := grpc.Dial(addr, grpc.WithInsecure())
	if err != nil {
		return errors.Trace(err)
	}
	defer conn.Close()

	client := debugpb.NewDebugClient(conn)

	// Recover
	recoverReq := &debugpb.RecoverFailPointRequest{
		Name: fail,
	}
	Infof("FailNemesis recovers service %s(%s):%s, fp: %+v", n.kind, n.name, addr, recoverReq)

	_, err = client.RecoverFailPoint(ctx, recoverReq)
	return errors.Trace(err)
}

// Operation defines the Daemon operation
type Operation struct {
	ContainerIP      string `json:"container_ip"`
	ContainerID      string `json:"container_id"`
	Action           string `json:"action"`
	Rate             string `json:"rate"`
	Operation        string `json:"operation"`
	Method           string `json:"method"`
	DirName          string `json:"dirname"`
	HostIP           string `json:"host_ip"`
	Egress           int    `json:"egress"`
	Ingress          int    `json:"ingress"`
	Latency          int    `json:"latency"`
	LatencyVariation int    `json:"latency_variation"`
	Methods          string `json:"methods"`
	Errno            uint32 `json:"errno"`
	Random           bool   `json:"random"`
	Pct              uint32 `json:"pct"`
	Path             string `json:"path"`
	Delay            uint32 `json:"delay"`
}

func (n *Node) daemonAction(operation string, port int, data []byte) error {
	n.Lock()
	defer n.Unlock()

	url := fmt.Sprintf("http://%s:%d/%s/%s", n.podInfo.NodeIP, port, daemonPrefix, operation)
	return errors.Trace(daemonAction(n.ctx, url, data))
}

func genNodes(ctx context.Context, pods []*PodInfo, kind string) []*Node {
	var (
		nodes []*Node
		err   error
	)
	for _, p := range pods {
		if p.IP == "" || p.ServicePort == 0 || p.AgentPort == 0 {
			continue
		}

		// use ip:agentPort to create agent client for nemesis to use.
		// agent controls the state of tidb、pd、tikv.
		node := &Node{
			cli:     agClient.NewClient(fmt.Sprintf("%s:%d", p.IP, p.AgentPort)),
			name:    p.Name,
			kind:    kind,
			podInfo: p,
			ctx:     ctx,
		}
		node.State, err = node.cli.ProcessState(kind)
		if err != nil {
			node.State = daemon.ProcStatUnknown
		}
		nodes = append(nodes, node)
	}

	return nodes
}