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

package ants

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

//Transfer TCP请求的报文协议解析、分发，响应
type Transfer struct {
	tcpConn      *net.TCPConn
	request      func() *Request
	outReadTime  int64
	outWriteTime int64
	user         string
	password     string
	isNewRequest bool
	isPconnect   bool
	closeSignal  bool
	clientID     string
	lock         sync.RWMutex
	writeLock    chan []byte
	clusterCheck LeaderCheckInterface
}

//Transport 接受请求，分发处理
func (that *Transfer) Transport() {
	that.isPconnect = false
	that.closeSignal = false
	that.isNewRequest = true
	that.writeLock = make(chan []byte, 1)
	_requestTime := 0
	defer func() {
		if err := recover(); err != nil {
			that.CloseSign()
			return
		}
	}()
	//
	for {
		if that.Closed() {
			return
		}
		if _err := that.getRequest(that.isNewRequest); _err != nil {
			that.CloseSign()
			if _err.Error() == "EOF" {
				log.Debugf("conn[%s] received close signal", that.clientID)
			}
			return
		}
		_requestTime++
	}
}

//getRequest 接收报文
func (that *Transfer) getRequest(isAuth bool) error {
	if that.Closed() {
		return nil
	}
	//设置超时
	_readDead := time.Duration(that.outReadTime) * time.Second
	if isAuth {
		//验证账号、密码，200毫秒快速释放
		_readDead = time.Duration(200) * time.Millisecond
	}
	if _err := that.tcpConn.SetReadDeadline(time.Now().Add(_readDead)); _err != nil {
		return _err
	}
	defer func() {
		that.tcpConn.SetReadDeadline(time.Time{})
	}()
	//验证密钥
	if isAuth {
		if isAuth {
			_cmdType, _len, _err := that.readHeader()
			if _err != nil {
				return _err
			}
			_cmdTypeSlice := strings.Split(_cmdType, "|")
			if _cmdTypeSlice[0] != "client-auth" {
				log.Error("conn connect fail 001")
				return errors.New("auth fail")
			}
			if _len < 20 {
				log.Error("conn connect fail 002")
				return errors.New("auth fail2")
			}
			_body, _err := that.read(_len)
			if _err != nil {
				log.Error("conn connect fail 003")
				return _err
			}
			_clientID := that.trim(_body[0:20])
			_auth := that.trim(_body[20:])
			if _auth != that.user+that.password {
				if _err := that.WriteByRequestID(strings.Join(_cmdTypeSlice, "|"), 401, []byte{}); _err != nil {
					log.Errorf("conn[%s] connect fail, response write error(%s)", _clientID, _err.Error())
					return _err
				}
				log.Errorf("conn[%s] connect fail, token error", _clientID)
				return errors.New("connect fail,token error! ")
			}
			if _err := that.WriteByRequestID(strings.Join(_cmdTypeSlice, "|"), 200, []byte{}); _err != nil {
				return _err
			}
			that.clientID = _clientID
			that.isNewRequest = false
			log.Debugf("conn[%s] connect success", _clientID)
			return nil
		}
	}
	//接受请求
	_cmdType, _len, _err := that.readHeader()
	if _err != nil {
		return _err
	}
	_cmdTypeSlice := strings.Split(_cmdType, "|")
	switch _cmdTypeSlice[0] {
	case "client-ping":
		log.Debugf("conn[%s] ping", that.clientID)
		if _len > 0 {
			if _, _err := that.read(_len); _err != nil {
				log.Debugf("conn[%s] ping, read error(%s)", that.clientID, _err.Error())
				return _err
			}
		}
	case "leader":
		log.Debugf("conn[%s] leader", that.clientID)
		if _len > 0 {
			if _, _err := that.read(_len); _err != nil {
				log.Debugf("conn[%s] isLeader, read error(%s)", that.clientID, _err.Error())
				return _err
			}
		}
		if that.clusterCheck != nil {
			//_code := 404
			_leaderID := that.clusterCheck.Leader()
			if _leaderID == "" {
				_leaderID = "0"
			}
			if _err := that.WriteByRequestID(strings.Join(_cmdTypeSlice, "|"), 200, []byte(_leaderID)); _err != nil {
				return _err
			}
		}
	case "isLeader":
		log.Debugf("conn[%s] isLeader", that.clientID)
		if _len > 0 {
			if _, _err := that.read(_len); _err != nil {
				log.Debugf("conn[%s] isLeader, read error(%s)", that.clientID, _err.Error())
				return _err
			}
		}
		if that.clusterCheck != nil {
			_code := 404
			if that.clusterCheck.LeaderCheck() {
				_code = 200
			}
			if _err := that.WriteByRequestID(strings.Join(_cmdTypeSlice, "|"), _code, []byte{}); _err != nil {
				return _err
			}
		}
	case "req":
		if _len > 0 {
			_body, _err := that.read(_len)
			if _err != nil {
				log.Errorf("conn[%s] req[%s], read error(%s)", that.clientID, _cmdType, _err.Error())
				return _err
			}
			//
			if that.clusterCheck != nil {
				if !that.clusterCheck.LeaderCheck() {
					if _err := that.WriteByRequestID(strings.Join(_cmdTypeSlice, "|"), 404, []byte{}); _err != nil {
						log.Errorf("conn[%s] req[%s], response write error(%s)", that.clientID, _cmdType, _err.Error())
						return _err
					}
					return nil
				}
			}
			//解析请求
			go func() {
				defer func() {
					if _err := recover(); _err != nil {
						_code, _rs := that.createResponse(errors.New(fmt.Sprint(_err)))
						if err := that.WriteByRequestID(_cmdType, _code, _rs); err != nil {
							that.CloseSign()
						}
						log.Error("conn[%s] req[%s], recover error(%s)", that.clientID, _cmdType, fmt.Sprint(_err))
					}
				}()
				if _err := that.ParseRequest(_cmdType, _body); _err != nil {
					that.CloseSign()
					log.Error("conn[%s] req[%s], error(%s)", that.clientID, _cmdType, fmt.Sprint(_err))
				}
			}()
		}
	default:
		log.Warnf("conn[%s] %s, this cmd type illegal", that.clientID, _cmdTypeSlice[0])
		return errors.New("this cmd type illegal")
		//未知类型的数据
	}
	return nil
}

//readHeader 解析请求报文头，得到报文内容字节的长度
func (that *Transfer) readHeader() (string, int, error) {
	//解析头
	_headByte, _err := that.read(30)
	if _err != nil {
		return "", 0, _err
	}
	_cmdType := that.trim(_headByte[0:25])
	//
	_bodyLength := 0
	_cmdBodyLength := that.trim(_headByte[25:])
	if _cmdBodyLength != "" {
		_n, _err := strconv.Atoi(that.trim(_headByte[25:]))
		if _err != nil {
			return "", 0, _err
		}
		_bodyLength = _n
	}
	return _cmdType, _bodyLength, nil
}

func (that *Transfer) read(bodyLength int) ([]byte, error) {
	if bodyLength > 0 {
		return that.readByte([]byte{}, bodyLength, 1)
	}
	return []byte{}, nil
}

//readByte 递归接收完报文
func (that *Transfer) readByte(readByte []byte, length int, readTime int) ([]byte, error) {
	if that.Closed() {
		return readByte, errors.New(that.clientID + " client conn closed")
	}
	if length > 0 {
		_datainfo := make([]byte, length)
		if readTime > 1000 {
			return readByte, errors.New("read lost data more than 1000 times! ")
		}
		_n, _err := that.tcpConn.Read(_datainfo)
		if _err != nil {
			if _n > 0 {
				readByte = append(readByte, _datainfo[0:_n]...)
			}
			return readByte, _err
		}
		lostLen := length - _n
		if _n > 0 {
			readByte = append(readByte, _datainfo[0:_n]...)
		}
		if lostLen > 0 {
			readTime = readTime + 1
			return that.readByte(readByte, lostLen, readTime)
		}
		return readByte, nil
	}
	return readByte, nil
}

//ParseRequest 处理接受到的数据，并最终返回结果。
func (that *Transfer) ParseRequest(cmdType string, get []byte) error {
	_path := strings.TrimSpace(that.trim(get[0:60]))
	_params := get[60:len(get)]
	log.Debugf("conn[%s] req[%s] path[%s] params[%s] ", that.clientID, cmdType, _path, string(_params))
	//请求解析
	_request := that.request()
	if _err := _request.RouteMatch(_path, _params); _err != nil {
		log.Errorf("conn[%s] req[%s] path[%s], route match error(%s)", that.clientID, cmdType, _path, _err.Error())
		_code, _rs := that.createResponse(_err)
		if err := that.WriteByRequestID(cmdType, _code, _rs); err != nil {
			return _err
		}
		return nil
	}
	//
	_response := _request.Handler()
	//请求转发
	_code, _rs := that.createResponse(_response)
	log.Debugf("conn[%s] req[%s] path[%s] params[%s], %d|%s ", that.clientID, cmdType, _path, string(_params), _code, string(_rs))
	if _err := that.WriteByRequestID(cmdType, _code, _rs); _err != nil {
		return _err
	}
	return nil
}

//WriteResponse 返回结果
func (that *Transfer) createResponse(data interface{}) (int, []byte) {
	_json := struct {
		Code    int         `json:"code"`
		Message string      `json:"message"`
		Data    interface{} `json:"data"`
	}{
		Code:    200,
		Message: "Success",
	}
	switch _data := data.(type) {
	case utils.UtilsException:
		_json.Code = utils.ToInt(_data.GetCode())
		_json.Message = _data.GetMessage()
	case error:
		_json.Code = 400
		_json.Message = _data.Error()
	case string:
		_json.Data = _data
	default:
		_json.Data = _data
	}
	//
	_headerCode := 200
	////特殊错误码
	switch _json.Code {
	case 404:
		//not leader
		_headerCode = 404
	}
	_send, _ := json.Marshal(_json)
	return _headerCode, _send
}

func (that *Transfer) WriteByRequestID(requestID string, code int, cmdByte []byte) error {
	_cmdType := []string{
		requestID,
		utils.ToString(code),
	}
	return that.Write(strings.Join(_cmdType, "|"), cmdByte)
}

//Write 追加头
func (that *Transfer) Write(cmdType string, cmdByte []byte) error {
	if that.Closed() {
		return errors.New(that.clientID + " client conn closed")
	}
	_sendByte := that.fiexLeninfo(cmdType, 25)
	_sendByte = append(_sendByte, that.fiexLeninfo(strconv.Itoa(len(cmdByte)), 5)...)
	if len(cmdByte) > 0 {
		_sendByte = append(_sendByte, cmdByte...)
	}
	//阻塞上锁
	that.writeLock <- []byte{}
	defer func() {
		//释放锁
		<-that.writeLock
		that.tcpConn.SetWriteDeadline(time.Time{})
	}()
	//设置超时
	if _err := that.tcpConn.SetWriteDeadline(time.Now().Add(time.Duration(3) * time.Second)); _err != nil {
		//loger.Error("that.conn.SetWriteDeadline(time.Now().Add(time.Duration(3) * time.Second)) return err:" + _err.Error())
		return _err
	}
	if _, _err := that.tcpConn.Write(_sendByte); _err != nil {
		return _err
	}
	return nil
}

//返回的“信息头”---包含将要返回多少字节给客户端
func (that *Transfer) fiexLeninfo(str string, l int) []byte {
	_minsend := make([]byte, l)
	_first := append([]byte(str), _minsend...)
	_first = _first[0:l]
	return _first
}

func (that *Transfer) trim(b []byte) string {
	_emptyByte := make([]byte, 1)
	return strings.TrimSpace(strings.Trim(string(b), string(_emptyByte)))
}

func (that *Transfer) Closed() bool {
	that.lock.RLock()
	defer that.lock.RUnlock()
	return that.closeSignal
}

func (that *Transfer) CloseSign() {
	that.lock.Lock()
	defer that.lock.Unlock()
	that.closeSignal = true
}

func (that *Transfer) Close() {
	that.tcpConn.Close()
	that.tcpConn = nil
	that.request = nil
	that.clusterCheck = nil
	log.Debugf("conn[%s] will close", that.clientID)
}
