/**
 * @desc		TCP框架
 * ********************************************
 * @author		unphp <unphp@qq.com>
 * @date		2021-12-17
 * @copyright	ants 0.1
 * ********************************************
 */

package ants

import (
	"PPOSUtils/utils"
	"encoding/json"
	"errors"
	"strings"
	"sync"
	"time"
)

func NewEngine(addressSlice, auth string, staticWorker, maxWorker int64) *Engine {
	_addressSlice := strings.Split(addressSlice, ",")
	_auth := strings.Split(auth, "@")
	_engine := &Engine{
		leaderCheckSign: make(chan string, 1),
		nodes:           map[string]*Node{},
	}
	for _, _v := range _addressSlice {
		_addressInfo := strings.Split(_v, "/")
		if len(_addressInfo) != 2 {
			log.Panic("node config error")
			panic("node config error!")
		}
		_nodeID := _addressInfo[0]
		_address := strings.Split(_addressInfo[1], ":")
		if len(_address) != 2 {
			log.Panic("node config error")
			panic("node config error")
		}
		//
		_engine.nodes[_nodeID] = &Node{
			nodeID:           _nodeID,
			host:             _address[0],
			port:             utils.ToInt(_address[1]),
			user:             _auth[0],
			password:         _auth[1],
			clientMp:         map[string]*ClientConn{},
			clientWorkerPool: make(chan string, staticWorker),
			getClientID: func(id string) string {
				return _engine.newClientID(id)
			},
		}
	}
	return _engine
}

//Engine ...
type Engine struct {
	nodes           map[string]*Node
	leaderNode      string
	leaderCheckSign chan string
	clientID        int64
	clientLock      sync.RWMutex
}

//ListenLeader leader监听
func (that *Engine) Listen() {
	//初始化静态链接池
	go func() {
		_initState := map[string]bool{}
		for {
			_pass := true
			for _k, _v := range that.nodes {
				_, _ok := _initState[_k]
				if !_ok {
					_initState[_k] = false
				}
				if _initState[_k] {
					continue
				}
				if _v.initStaticWorker() {
					_initState[_k] = true
				}
			}
			for _, _ok := range _initState {
				if !_ok {
					_pass = false
				}
			}
			if _pass {
				break
			}
		}
	}()
	//leader监听
	go func() {
		for {
			that.voteLeader("0")
			time.Sleep(time.Duration(1000) * time.Millisecond)
		}
	}()
}

//voteLeaderSign 发送获取选举leader信号
func (that *Engine) voteLeaderSign(excludeNodes ...string) {
	_excludeNode := "0"
	if len(excludeNodes) > 0 {
		_excludeNode = excludeNodes[0]
	}
	go that.voteLeader(_excludeNode)
}

//voteLeader 获取选举的leader
func (that *Engine) voteLeader(excludeNode string) {
	select {
	case that.leaderCheckSign <- excludeNode:
		func() {
			defer func() {
				<-that.leaderCheckSign
				if _err := recover(); _err != nil {
					log.Error("[voteLeader] error", _err)
				}
			}()
			log.Infof("[Engine.voteLeader] excludeNode:%s", excludeNode)
			_excludeSlice := []string{}
			if excludeNode != "0" {
				//that.delLeader(excludeNode)
				_excludeSlice = append(_excludeSlice, excludeNode)
			}
			_v := that.checkLoop(_excludeSlice...)
			if _v != "0" {
				that.setLeader(_v)
			}
		}()
	default:

	}
}

//checkLoop ...
func (that *Engine) checkLoop(excludeNodes ...string) string {
	_i := 0
	excludeNode := ""
	if len(excludeNodes) > 0 {
		excludeNode = excludeNodes[0]
	}
	for {
		_sleepTime := 50
		if _i > 1 {
			excludeNode = ""
		}
		if _i > 20 {
			_sleepTime = 100
		}
		if _i > 40 {
			_sleepTime = 200
		}
		if _i > 100 {
			return "0"
		}
		for _k, _v := range that.nodes {
			if _k == excludeNode {
				continue
			}
			_id, _err := _v.Leader()
			if _err != nil {
				continue
			}
			if _id == "0" {
				continue
			}
			return _id
		}
		time.Sleep(time.Duration(_sleepTime) * time.Millisecond)
		_i++
	}
}

//NewRequest 节点请求
func (that *Engine) NewRequest(path string, params interface{}) ([]byte, error) {
	var _params []byte
	switch _data := params.(type) {
	case []byte:
		_params = _data
	default:
		_params, _ = json.Marshal(_data)
	}
	_rs, _err := that.request(path, _params, 1, 20)
	return _rs, _err
}

func (that *Engine) request(path string, params []byte, times, agentTimes int) ([]byte, error) {
	if times > agentTimes {
		log.Error("node not fount leader")
		return nil, errors.New("[" + utils.TimeStr("Y-m-d h:i:s") + "] node not fount leader")
	}
	//获取leader
	_leaderNodeID := that.getLeader()
	if _leaderNodeID == "" {
		//没有找到leader，重新选举
		log.Info("[001] not fount leader, will check nodes leader")
		that.voteLeaderSign()
		time.Sleep(time.Duration(200) * time.Millisecond)
		return that.request(path, params, times+1, agentTimes)
	}
	_leaderNode := that.nodes[_leaderNodeID]
	log.Infof("get leader: %s", _leaderNodeID)
	//从链接池中，获取可用的链接
	_session, _err := _leaderNode.NewSession()
	if _err != nil {
		//网络异常，重新获取leader
		log.Info("[002] not fount leader, will check nodes leader")
		that.voteLeaderSign(_leaderNodeID)
		time.Sleep(time.Duration(200) * time.Millisecond)
		return that.request(path, params, times+1, agentTimes)
	}
	defer _session.Close()

	//检测链接是否正常
	if _err := _session.Ping(); _err != nil {
		//网络异常，重新获取leader
		log.Info("[003] leader conn ping fail, will check nodes leader")
		that.voteLeaderSign(_leaderNodeID)
		time.Sleep(time.Duration(200) * time.Millisecond)
		return that.request(path, params, times+1, agentTimes)
	}
	//业务请求
	_code, _data, _err := _session.NewRequest(path, params)
	if _code != 200 {
		if _code == 403 {
			//链接已关闭
			log.Info("[004] leader conn closed, will check nodes leader")
			if _err != nil {
				log.Errorf("error:%s", _err.Error())
			}
			time.Sleep(time.Duration(200) * time.Millisecond)
			return that.request(path, params, times+1, agentTimes)
		}
		if _code == 404 {
			//业务请求404，代表not leader
			log.Infof("[005] leader(%s) is change, will check nodes leader", _session.GetWorkerID())
			that.voteLeaderSign(_leaderNodeID)
			time.Sleep(time.Duration(200) * time.Millisecond)
			return that.request(path, params, times+1, agentTimes)
		}
		if _err != nil {
			log.Errorf("node(%s) request error: %s", _session.GetWorkerID(), _err.Error())
			return nil, _err
		}
		//
		log.Errorf("node(%s) request error: %s", _session.GetWorkerID(), "未知错误")
		return nil, errors.New("未知错误")
	}
	return _data, nil
}

func (that *Engine) getLeader() string {
	that.clientLock.RLock()
	defer that.clientLock.RUnlock()
	return that.leaderNode
}

func (that *Engine) delLeader(leaderNode string) {
	that.clientLock.Lock()
	defer that.clientLock.Unlock()
	if that.leaderNode == leaderNode {
		that.leaderNode = ""
	}
}

func (that *Engine) setLeader(leaderNode string) {
	that.clientLock.Lock()
	defer that.clientLock.Unlock()
	that.leaderNode = leaderNode
}

func (that *Engine) newClientID(nodeID string) string {
	that.clientLock.Lock()
	defer that.clientLock.Unlock()
	that.clientID = that.clientID + 1
	_slice := []string{
		nodeID,
		utils.ToString(that.clientID),
	}
	return strings.Join(_slice, "-")
}
