package serf

import (
	"basic.com/valib/logger.git"
	"basic.com/valib/serf.git/serf"
	"encoding/json"
	"github.com/golang/protobuf/proto"
	"github.com/hashicorp/memberlist"
	"github.com/satori/go.uuid"
	"path/filepath"
	"reflect"
	"runtime"
	"strconv"
	"strings"
	"time"
	"vamicro/config"
	"vamicro/system-service/bhome_msg_dev"
)

type RpcHandle func(arg RpcParamTopic) ([]serf.NodeResponse, error)

var rpcHandlers map[string]RpcHandle

func init() {
	rpcHandlers = make(map[string]RpcHandle)
}

// RegisterRpcHandles
func RegisterRpcHandles(fs ...RpcHandle) {
	for _, f := range fs {
		name := runtime.FuncForPC(reflect.ValueOf(f).Pointer()).Name() //absolute package path name eg: a.b.c.d.FuncName
		nameEnd := filepath.Ext(name)
		name = strings.TrimPrefix(nameEnd, ".")
		if _, ok := rpcHandlers[name]; !ok {
			rpcHandlers[name] = f
		}
	}
}

/*****************************UserEvent***************************************/
func HandleUserEventSyncSql(ev serf.UserEvent) {
	logger.Info("receive a UserEventSyncSql event")
	var sqlUe SqlUserEvent
	err := json.Unmarshal(ev.Payload, &sqlUe)
	if err != nil {
		logger.Error("sqlUe unmarshal err:", err)
		return
	}

	logger.Info("ev.LTime:", ev.LTime, "owner:", sqlUe.Owner, "sql:", sqlUe.Sql)
	if sqlUe.Owner != config.Server.AnalyServerId {
		go func() {
			flag, e := executeSqlByGorm(sqlUe.Sql)
			logger.Info("ev.LTime:", ev.LTime, "userEvent exec ", sqlUe.Sql, ",Result:", flag, ", err:", e)
			logLT := strconv.Itoa(int(ev.LTime))
			logT := time.Now().Format("2006-01-02 15:04:05")
			logSql := strings.ReplaceAll(strings.Join(sqlUe.Sql, ";"), "'", "''")
			logResult := "0"
			if flag {
				logResult = "1"
			}
			logErr := ""
			if e != nil {
				logErr = e.Error()
			}
			executeSqlByGorm([]string{"insert into sql_sync_his(`id`,`lTime`,`createTime`,`sql`,`from`,`result`,`err`) values ('" + uuid.NewV4().String() + "','" + logLT + "','" + logT + "','" + logSql + "','" + sqlUe.Owner + "'," + logResult + ",'" + logErr + "')"})
		}()
	}
}

func HandleUserEventSyncDbTablePersonCache(ev serf.UserEvent) {
	logger.Info("LTime:", ev.LTime, ",ev.Payload.len:", len(ev.Payload))
	SyncDbTablePersonCacheChan <- ev.Payload
}

func HandleUserEventSyncVirtualIp(ev serf.UserEvent) {
	logger.Info("LTime:", ev.LTime, " Recevie virtualIp change")
	SyncVirtualIpChan <- ev.Payload
}

//收到其它节点主动将注册中心的所有topic通知到集群中
func HandleSyncRegisterInfo(ev serf.UserEvent) {
	logger.Debug("HandleSyncRegisterInfo")
	var si bhome_msg_dev.MsgDevRegisterInfo
	if err := proto.Unmarshal(ev.Payload, &si); err == nil {
		logger.Debug("HandleSyncRegisterInfo si.DevId:", string(si.DevId), " config.Server.AnalyServerId:", config.Server.AnalyServerId)
		if string(si.DevId) != config.Server.AnalyServerId {
			compareRPool(&si)
		}
	} else {
		logger.Error("HandleSyncRegisterInfo unmarshal err:", err)
	}
}

func HandleDataSystemSerfSub(ev serf.UserEvent) {
	h := GetBusHandle()
	if h == nil {
		logger.Error("HandleDataSystemSerfSub bus handle is nil")
		return
	}
	err := h.Publish(DataSystemSerfSubscribe, ev.Payload)
	if err != nil {
		logger.Error("HandleDataSystemSerfSub pub err:", err)
	}
}

/*****************************Query***************************************/
func HandleQueryEventUpdateDBData(ev *serf.Query) {
	logger.Info("receive QueryEventUpdateDBData, current node:", config.Server.AnalyServerId)
	var fromP QueryTableDataParam
	err := json.Unmarshal(ev.Payload, &fromP)
	if err != nil {
		logger.Error("Query tableNames unmarshal err")
		if err := ev.Respond([]byte("request unmarshal err")); err != nil {
			logger.Error("query.Respond err: %s\n", err)
			return
		}

		return
	}
	logger.Info("Query tableNames:", fromP.Tables)
	datas, err := DumpTables(fromP.Tables)
	if err != nil {
		logger.Error("queryByGorm err:", err)
		if err := ev.Respond([]byte("queryByGorm err")); err != nil {
			logger.Error("query.Respond err: %s\n", err)
			return
		}
		return
	}
	bytesReturn, err := json.Marshal(datas)
	logger.Info("results.len: ", len(bytesReturn))

	var targetNode *memberlist.Node
	nodes := Agent.Serf().Memberlist().Members()
	if nodes != nil && len(nodes) > 0 {
		for _, n := range nodes {
			if n.Name == fromP.From {
				targetNode = n
				break
			}
		}
	}
	logger.Debug("targetNode:", targetNode.Name)
	if targetNode != nil {
		go func() {
			addr := targetNode.Addr.String() + ":" + strconv.Itoa(TcpTransportPort)
			sendErr := rawSendTcpMsg(addr, bytesReturn)

			logLT := strconv.Itoa(int(ev.LTime))
			logT := time.Now().Format("2006-01-02 15:04:05")
			logSql := strings.ReplaceAll("QueryEventUpdateDBData from "+targetNode.Name, "'", "''")
			logResult := "0"
			logErr := ""
			if sendErr == nil {
				logResult = "1"
				logger.Debug("sendToTcp success")
			} else {
				logErr = sendErr.Error()
				logger.Debug("sendToTcp err:", sendErr)
			}

			executeSqlByGorm([]string{"insert into sql_sync_his(`id`,`lTime`,`createTime`,`sql`,`from`,`result`,`err`) values ('" + uuid.NewV4().String() + "','" + logLT + "','" + logT + "','" + logSql + "','" + targetNode.Name + "'," + logResult + ",'" + logErr + "')"})
		}()
	} else {
		logger.Debug("targetNode is nil")
	}
}

//处理其他的一些query请求
func HandleOtherQuery(ev *serf.Query) {
	var reqBody RequestSerfTopicMsg
	var ret []byte
	if err := json.Unmarshal(ev.Payload, &reqBody); err != nil {
		ret = []byte(err.Error())
	} else {
		if err, data := QueryLocalProc(reqBody); err != nil {
			ret = []byte(err.Error())
		} else {
			b, e := json.Marshal(data)
			if e != nil {
				ret = []byte(e.Error())
			} else {
				ret = b
			}
		}
	}

	if err := ev.Respond(ret); err != nil {
		logger.Debug("HandleOtherQuery err:", err)
		return
	}
}

func HandleQueryRpc(ev *serf.Query) {
	var ret []byte
	var arg RpcParamTopic
	err := json.Unmarshal(ev.Payload, &arg)
	if err == nil {
		if f, ok := rpcHandlers[arg.Topic]; ok {
			resp, e := f(arg)
			if e == nil {
				if data, me := json.Marshal(resp); me == nil {
					ret = data
				} else {
					logger.Debug("marshal resp err:", e)
				}
			} else {
				logger.Debug("call f err:", e)
			}
		} else {
			logger.Debug("rpcHandlers not contains topic:", arg.Topic)
		}
	} else {
		logger.Debug("unmarshal RpcParamTopic err:", err)
	}
	if rErr := ev.Respond(ret); rErr != nil {
		logger.Debug("HandleQueryRpc err:", rErr)
	}
}

func HandleEventMemberLeave(ev serf.MemberEvent) {
	if ev.Members != nil && len(ev.Members) == 1 {
		leaveMember := ev.Members[0]
		leaveSql := "update cluster_node set isDelete=1 where node_id='" + leaveMember.Name + "'"
		flag, e := executeSqlByGorm([]string{leaveSql})

		logger.Info("EventMemberLeave,current Members:", ev.Members)
		logLT := ""
		logT := time.Now().Format("2006-01-02 15:04:05")
		logSql := strings.ReplaceAll(leaveSql, "'", "''")
		logResult := "0"
		if flag {
			logResult = "1"
		}
		logErr := ""
		if e != nil {
			logErr = e.Error()
		}
		executeSqlByGorm([]string{"insert into sql_sync_his(`id`,`lTime`,`createTime`,`sql`,`from`,`result`,`err`) values ('" + uuid.NewV4().String() + "','" + logLT + "','" + logT + "','" + logSql + "','" + leaveMember.Name + "'," + logResult + ",'" + logErr + "')"})
	}
}

func HandleEventMemberJoin(ev serf.MemberEvent) {
	if ev.Members != nil && len(ev.Members) == 1 {
		leaveMember := ev.Members[0]
		joinSql := "update cluster_node set isDelete=0 where node_id='" + leaveMember.Name + "'"
		flag, e := executeSqlByGorm([]string{joinSql})

		logger.Info("EventMemberJoin,current Members:", ev.Members)
		logLT := ""
		logT := time.Now().Format("2006-01-02 15:04:05")
		logSql := strings.ReplaceAll(joinSql, "'", "''")
		logResult := "0"
		if flag {
			logResult = "1"
		}
		logErr := ""
		if e != nil {
			logErr = e.Error()
		}
		executeSqlByGorm([]string{"insert into sql_sync_his(`id`,`lTime`,`createTime`,`sql`,`from`,`result`,`err`) values ('" + uuid.NewV4().String() + "','" + logLT + "','" + logT + "','" + logSql + "','" + leaveMember.Name + "'," + logResult + ",'" + logErr + "')"})
	}
}