package sdk

import (
	"taskpubsub/tasktag"
	"taskpubsub/util"

	"github.com/gogo/protobuf/proto"

	"basic.com/pubsub/protomsg.git"
	"basic.com/valib/deliver.git"
	"basic.com/valib/logger.git"
)

const (
	postPush = "_1"
	postPull = "_2"
)

var SocketManage = make(map[string]util.SocketContext)
var SdkMap = make(map[string]chan protomsg.SdkMessage)

var shm bool = false

var innerRecvTopic = []string{
	"facedetect-sdk-no-track", //to sdk-no-track 以图搜图
}

var innerSendTopic = []string{
	"facedetect-sdk-no-track",      //to sdk-no-track 以图搜图
	"virtual-faceextract-sdk-pull", //to web 以图搜图
}

func initInnerTopic() {
	// for _, sendTopic := range innerSendTopic {
	// 	createSdkSendServerAndListen(sendTopic)
	// }

	// for _, recvTopic := range innerRecvTopic {
	// 	createSdkRecvServerAndListen(recvTopic)
	// }
}

func Init(useShm bool) {

	shm = useShm

	logger.Info("============= init sdk info =====================")
	for _, sdkid := range util.Sdklist { // 创建sdk server
		createSdkTopicAndServer(sdkid)
		logger.Info()
	}

	// 手动输入的主题
	initInnerTopic()

	// es
	SdkMap["es"] = make(chan protomsg.SdkMessage)
	logger.Info("create es channel: ")
	go DealEsTopic()

	go autoUpdateSdk(util.Sdkflag)
}

func createSdkTopicAndServer(sdkid string) {
	createSdkSendServerAndListen(sdkid)
	createSdkRecvServerAndListen(sdkid)
}

func deleteSdkTopicAndServer(id string) {
	if _, isExist := SdkMap[id]; isExist { //存在
		close(SdkMap[id])
		delete(SdkMap, id)
		logger.Info("删除主题 sdk: ", id)
	}

	if _, isExist := SocketManage[id]; isExist { //存在
		SocketManage[id].Cancel()
		delete(SocketManage, id)
		logger.Info("删除server sdk: ", id)
	}
}

func createSdkSendServerAndListen(id string) {
	if _, isExist := SdkMap[id]; !isExist { //不存在
		SdkMap[id] = make(chan protomsg.SdkMessage)
		logger.Info("create", id)
	}

	url := "ipc:///tmp/" + id + postPush + ".ipc"
	m := deliver.PushPull
	if shm {
		m = deliver.Shm
		url = id + postPush
	}

	logger.Info("SDK URL: ", url)

	socket, err := util.NewSocketListen(int(m), url, shm)
	if err != nil {
		delete(SdkMap, id)
		logger.Error(id, "create socket error!")
		return
	}
	SocketManage[id] = socket

	go Send(id, socket, SdkMap[id])
}

func createSdkRecvServerAndListen(id string) {
	if _, isExist := SdkMap[id]; !isExist { //不存在
		SdkMap[id] = make(chan protomsg.SdkMessage)
		logger.Info("create", id)
	}

	url := "ipc:///tmp/" + id + postPull + ".ipc"
	m := deliver.PushPull
	if shm {
		m = deliver.Shm
		url = id + postPull

	}

	socket, err := util.NewSocketListen(int(m), url, shm)

	if err != nil {
		delete(SdkMap, id)
		logger.Error(id, "create socket error!")
		return
	}
	SocketManage[id] = socket

	go Recv(socket)
}

//单独处理   es 主题的情况
func DealEsTopic() {
	for {
		select {
		case <-SdkMap["es"]:
			//logger.Info("es finanl sdk!")
		}
	}
}

//动态处理
func autoUpdateSdk(sdkflag chan bool) {

	for _ = range sdkflag {
		logger.Info("test autodelsdk")
		var oldSdkList []string
		for key := range SdkMap {
			oldSdkList = append(oldSdkList, key)
		}

		newSdkList := util.Sdklist

		// 手动添加的全部加上
		// for _, sendTopic := range innerSendTopic {
		// 	newSdkList = append(newSdkList, sendTopic)
		// }
		// for _, recvTopic := range innerRecvTopic {
		// 	newSdkList = append(newSdkList, recvTopic)
		// }

		sdkListUpdate := util.Difference(oldSdkList, newSdkList)
		logger.Info(sdkListUpdate)

		for key, op := range sdkListUpdate {
			if op == "add" {
				createSdkTopicAndServer(key)
			} else {
				deleteSdkTopicAndServer(key)
			}
		}
	}
}

//sdk数据 加工器
func ToSdkMsg(cid string, caddr string, taskid string, data []byte) protomsg.SdkMessage {
	var sdkmsg = protomsg.SdkMessage{}
	sdkmsg.Cid = cid
	sdkmsg.Caddr = caddr
	if val, ok := tasktag.TaskLabelMap.Load(taskid); !ok {
		sdkmsg.Tasklab = nil
		return sdkmsg
	} else {
		sdkmsg.Tasklab = val.(*protomsg.TaskLabel)
		sdkmsg.Data = data
	}
	return sdkmsg
}

var logShouldUntil = 0

//sdk数据分发器
func GetSdkSendTopic(sdkmsg protomsg.SdkMessage) (sendTopic string) {
	if int(sdkmsg.Tasklab.Index) < len(sdkmsg.Tasklab.Sdkinfos) {
		sendTopic = sdkmsg.Tasklab.Sdkinfos[sdkmsg.Tasklab.Index].Ipcid

		logShouldUntil++
		if logShouldUntil > 68 {
			logShouldUntil = 0
			logger.Info("=========分发的主题=========")
			logger.Info("分发的主题:", sendTopic, "!Sdktype：", sdkmsg.Tasklab.Sdkinfos[sdkmsg.Tasklab.Index].Sdktype)
		}
	} else {
		sendTopic = "es"
	}

	return sendTopic
}

func Recv(socket util.SocketContext) {

	tryCount := 0

	var repsdkmsg = protomsg.SdkMessage{}
	for {
		select {
		case <-socket.Context.Done():
			socket.Sock.Close()

			logger.Info("socket close")
			return
		default:
			if msg, err := socket.Sock.Recv(); err != nil {

				// tryCount++
				// socket = util.MaybeRestartSocket(socket, &tryCount)

				if socket.UseSHM {
					if tryCount > util.ShmMaxTryCount {
						socket.Sock.Close()
						socket, err = util.NewSocketListen(socket.Mode, socket.URL, socket.UseSHM)
						logger.Info("=========SDK RECV SHM ERROR=========")
						logger.Info("SDK RECV FROM:", socket.URL, " ERROR ", util.ShmMaxTryCount, " TIMES SO RESTART IT")
						tryCount = 0
						continue
					}
					tryCount++
				}
				continue
			} else {
				err = proto.Unmarshal(msg, &repsdkmsg)
				if err != nil {
					logger.Error("unmarshal error: ", err)
					continue
				}
				repsdkmsg.Tasklab.Index++
				//调用计算函数， 分发给下一个主题
				nexttopic := GetSdkSendTopic(repsdkmsg)
				if "facedetect-sdk-no-track" == nexttopic || "virtual-faceextract-sdk-pull" == nexttopic {
					logger.Info("nexttopic:", nexttopic)
				}
				SdkMap[nexttopic] <- repsdkmsg
			}
		}
	}
}

func Send(sdkid string, socket util.SocketContext, in chan protomsg.SdkMessage) {

	tryCount := 0

	for {
		select {
		case <-socket.Context.Done():
			socket.Sock.Close()

			logger.Info("socket is close")
			return
		case v, ok := <-in:
			if ok {
				data, err := v.Marshal()
				if err != nil {
					logger.Error("proto marshal error ", err)
					continue
				}

				if err := socket.Sock.Send(data); err != nil {
					if "facedetect-sdk-no-track" == sdkid || "virtual-faceextract-sdk-pull" == sdkid {
						logger.Error("failed send:sdkid=", sdkid)
					}

					// tryCount++
					// socket = util.MaybeRestartSocket(socket, &tryCount)

					if socket.UseSHM {
						if tryCount > util.ShmMaxTryCount {
							socket.Sock.Close()
							socket, err = util.NewSocketListen(socket.Mode, socket.URL, socket.UseSHM)
							logger.Info("=========SDK SEND SHM ERROR=========")
							logger.Info("SDK SEND TO: ", socket.URL, " ERROR ", util.ShmMaxTryCount, " TIMES SO RESTART IT")
							tryCount = 0
							continue
						}
						tryCount++
					}

					continue
				}
				if "facedetect-sdk-no-track" == sdkid || "virtual-faceextract-sdk-pull" == sdkid {
					logger.Info(sdkid, " send success: ", len(data))
				}
			} else {
				logger.Debug(sdkid, " 主题关闭, 关闭send()")
				return
			}
		}
	}
}