package server

import (
	"net"
	"primesys_producer/src/kafkacommandproducer"
	"primesys_producer/src/kafkaconsumer"
	"primesys_producer/src/kafkaproducer"
	"primesys_producer/src/logger"
	"primesys_producer/src/protocol"
	"primesys_producer/src/protocol/heartbeat"
	lbsmultibase "primesys_producer/src/protocol/lbs_multibase"
	"primesys_producer/src/protocol/login"
	onlinecommand "primesys_producer/src/protocol/online_command"
	timecalibration "primesys_producer/src/protocol/time_calibration"
	"primesys_producer/src/server/frame"
	"primesys_producer/src/store"
	. "primesys_producer/src/types"
	"primesys_producer/src/utils"
	"runtime/debug"
	"strconv"
)

// open connection for devices to connect and Accept packets
func TCPServer(port string, kafkaTopic string, commandTopic string, producer *kafkaproducer.PacketProducer, commandProducer *kafkacommandproducer.CommandProducer, commandConsumer *kafkaconsumer.CommandConsumer) {

	ln, err := net.Listen("tcp", port)

	if err != nil {
		logger.Fatal("TCP Listener not started %v", err)
	}

	logger.Info("Listener started with port %s", port)

	// infinite loop to accept connections
	for {
		conn, err := ln.Accept()
		if err != nil {
			logger.Warn("Listener Connection has a connection error %v", err)
			continue
		}

		go handleConnection(conn, kafkaTopic, commandTopic, producer, commandProducer, commandConsumer)
	}
}

// handle connection with opening new go routines like creating a thread or another room for accepting the packets or data
func handleConnection(conn net.Conn, kafkaTopic string, commandTopic string, producer *kafkaproducer.PacketProducer, commandProducer *kafkacommandproducer.CommandProducer, commandConsumer *kafkaconsumer.CommandConsumer) {
	peer := conn.RemoteAddr()
	var imei int64

	logger.Info("peer= %s connected", peer)

	defer func() {
		// a bad packet from one device must not crash the whole server
		if r := recover(); r != nil {
			logger.Error("peer=%s panic recovered: %v\n%s", peer, r, debug.Stack())
		}
		if imei != 0 {
			store.DestoryImeiConn(imei, conn)
			logger.Info("imei=%d peer=%s disconnected", imei, peer)
		}
		conn.Close()
	}()

	var buf []byte
	readBuf := make([]byte, 4096)

	for {
		n, err := conn.Read(readBuf)
		if n > 0 {
			buf = append(buf, readBuf[:n]...)
			prevImei := imei
			buf, imei = drainFrame(conn, peer, buf, imei, kafkaTopic, commandTopic, producer, commandProducer, commandConsumer)
			if imei != 0 && imei != prevImei {
				store.StoreDeviceConnectionMap(imei, conn)
			}
		}
		if err != nil {
			logger.Warn("n = peer=%s read ended= %v", peer, err)
			return
		}
	}
}

// parse it and detect either this is complete incomplete or what kind of frame the connection have received
func drainFrame(conn net.Conn, peer net.Addr, buf []byte, imei int64, kafkaTopic string, commandTopic string, producer *kafkaproducer.PacketProducer, commandProducer *kafkacommandproducer.CommandProducer, commandConsumer *kafkaconsumer.CommandConsumer) ([]byte, int64) {
	for {
		fr, rest, status := frame.ExtractFrame(buf)
		switch status {
		case frame.Complete:
			logger.Info("peer = %s Complete Frame : %s", peer, utils.HexUpper(fr.Raw))
			imei = handleFrame(conn, peer, fr, imei, kafkaTopic, commandTopic, producer, commandProducer, commandConsumer)
			buf = rest
		case frame.Skip:
			skipped := len(buf) - len(rest)
			logger.Warn("Skipping unexpcted byte len : %d byte : %s", skipped, utils.HexUpper(buf[:skipped]))
			buf = rest
		case frame.Incomplete:
			logger.Warn("peer = %s Incomplete Frame : %d", peer, len(buf))
			return buf, imei
		}
	}
}

// this function will handle frame to write ack and save data to kafka
func handleFrame(conn net.Conn, peer net.Addr, frame frame.Frame, imei int64, kafkaTopic string, commandTopic string, kafkaProducer *kafkaproducer.PacketProducer, commandProducer *kafkacommandproducer.CommandProducer, commandConsumer *kafkaconsumer.CommandConsumer) int64 {

	logger.Debug("Protocol Number %02X", frame.ProtocolNumber())
	logger.Debug("Imei %d, %s", imei, utils.HexUpper(frame.Raw))

	switch frame.ProtocolNumber() {
	case ProtocolNumber(protocol.ProtoLogin):
		logger.Info("Login Packet start ===============================")
		data, err := login.DecodeLogin(frame)
		if err != nil {
			logger.Info("peer=%s login decode error: %v", peer, err)
			return imei
		}

		imei = data.Imei
		logger.Info("Login IMEI: %d serial: %d", imei, data.SerialNumber)

		ack := login.EncodeLoginAck(data.SerialNumber, frame.ProtocolNumber())

		if _, err := conn.Write(ack); err != nil {
			logger.Info("peer=%s frame out=%s", peer, utils.HexUpper(ack))
			return imei
		}

		request := kafkaproducer.ProducerRequest{
			Topic:    kafkaTopic,
			DeviceId: strconv.FormatInt(imei, 10),
			Packet: DevicePacket{
				ProtocolNumber: utils.HexByteUpper(byte(frame.ProtocolNumber())),
				Frame:          utils.HexUpper(frame.Raw),
				Raw:            frame.Raw,
				Ack:            utils.HexUpper(ack),
				PacketFrom:     "device",
				Time:           utils.ToEpochISTSecond(),
			},
		}

		if err := kafkaProducer.Send(request); err != nil {
			logger.Warn("peer = %s kafka Login publish error: %v", peer, err)
		}

		logger.Info("Login Packet end ===============================")

	case ProtocolNumber(protocol.ProtoHeartBeat):
		logger.Info("====== Heart beat packet 0x13 =====")
		data, err := heartbeat.DecodeHeartBeatSerial(frame)
		if err != nil {
			logger.Warn("peer=%s heartbeat decode error: %v", peer, err)
			return imei
		}

		ack := heartbeat.EncodeSerialACK(data.SerialNumber, frame.ProtocolNumber(), imei)

		if _, err := conn.Write(ack); err != nil {
			logger.Info("peer=%s heartBeat frame out=%s", peer, utils.HexUpper(ack))
			return imei
		}
		request := kafkaproducer.ProducerRequest{
			Topic:    kafkaTopic,
			DeviceId: strconv.FormatInt(imei, 10),
			Packet: DevicePacket{
				ProtocolNumber: utils.HexByteUpper(byte(frame.ProtocolNumber())),
				Frame:          utils.HexUpper(frame.Raw),
				Raw:            frame.Raw,
				Ack:            utils.HexUpper(ack),
				PacketFrom:     "device",
				Time:           utils.ToEpochISTSecond(),
			},
		}

		if err := kafkaProducer.Send(request); err != nil {
			logger.Warn("peer = %s kafka HeartBeat publish error: %v", peer, err)
			return imei
		}

		logger.Info("====== Heart beat packet 0x13 =====")
	case ProtocolNumber(protocol.ProtoHeartBeatPacket):
		logger.Info("HeartBeat Packet start ===============================")
		data, err := heartbeat.DecodeHeartBeatSerial(frame)
		if err != nil {
			logger.Warn("peer=%s heartbeat decode error: %v", peer, err)
			return imei
		}

		ack := heartbeat.EncodeSerialACK(data.SerialNumber, frame.ProtocolNumber(), imei)

		if _, err := conn.Write(ack); err != nil {
			logger.Info("peer=%s heartBeat frame out=%s", peer, utils.HexUpper(ack))
			return imei
		}

		request := kafkaproducer.ProducerRequest{
			Topic:    kafkaTopic,
			DeviceId: strconv.FormatInt(imei, 10),
			Packet: DevicePacket{
				ProtocolNumber: utils.HexByteUpper(byte(frame.ProtocolNumber())),
				Frame:          utils.HexUpper(frame.Raw),
				Raw:            frame.Raw,
				Ack:            utils.HexUpper(ack),
				PacketFrom:     "device",
				Time:           utils.ToEpochISTSecond(),
			},
		}

		if err := kafkaProducer.Send(request); err != nil {
			logger.Warn("peer = %s kafka HeartBeat publish error: %v", peer, err)
			return imei
		}
		logger.Info("HeartBeat Packet End ===============================")

	case ProtocolNumber(protocol.ProtoTimeCalibration):
		logger.Info("Time Calibration 0x1F start ===============================")
		data, err := timecalibration.DecodeTimeCalibration(frame)
		if err != nil {
			logger.Error("peer:= %s timecalibration frame out=%v", peer, err)
			return imei
		}

		ack := timecalibration.EncodeTimeCalibrationAck(data.SerialNumber, frame.ProtocolNumber(), data.LanguageExtendedByte)

		if _, err := conn.Write(ack); err != nil {
			logger.Error("peer=%s frame out = %s error = %v", peer, utils.HexUpper(ack), err)
			return imei
		}

		request := kafkaproducer.ProducerRequest{
			Topic:    kafkaTopic,
			DeviceId: strconv.FormatInt(imei, 10),
			Packet: DevicePacket{
				ProtocolNumber: utils.HexByteUpper(byte(frame.ProtocolNumber())),
				Frame:          utils.HexUpper(frame.Raw),
				Raw:            frame.Raw,
				Ack:            utils.HexUpper(ack),
				PacketFrom:     "device",
				Time:           utils.ToEpochISTSecond(),
			},
		}

		if err := kafkaProducer.Send(request); err != nil {
			logger.Error("perr = %s kafka Time Calibration packet error: %v", peer, err)
		}

		logger.Info("Time Calibration 0x1F end ===============================")

	case ProtocolNumber(protocol.ProtoGPSLocation):
		logger.Info("GPS Location Packet 0x22 start ===================================")

		request := kafkaproducer.ProducerRequest{
			Topic:    kafkaTopic,
			DeviceId: strconv.FormatInt(imei, 10),
			Packet: DevicePacket{
				ProtocolNumber: utils.HexByteUpper(byte(frame.ProtocolNumber())),
				Frame:          utils.HexUpper(frame.Raw),
				Raw:            frame.Raw,
				PacketFrom:     "device",
				Time:           utils.ToEpochISTSecond(),
			},
		}

		if err := kafkaProducer.Send(request); err != nil {
			logger.Error("Kafka GPS Location packet 0x22 error peer = %s error = %v", peer, err)
			return imei
		}

		logger.Info("GPS Location Packet 0x22 end ===================================")

	case ProtocolNumber(protocol.ProtoLocation2G):
		logger.Info("================= 2G GPS Location Packet ====================")
		logger.Info("Its 2g packet")
		request := kafkaproducer.ProducerRequest{
			Topic:    kafkaTopic,
			DeviceId: strconv.FormatInt(imei, 10),
			Packet: DevicePacket{
				ProtocolNumber: utils.HexByteUpper(byte(frame.ProtocolNumber())),
				Frame:          utils.HexUpper(frame.Raw),
				Raw:            frame.Raw,
				PacketFrom:     "device",
				Time:           utils.ToEpochISTSecond(),
			},
		}

		if err := kafkaProducer.Send(request); err != nil {
			logger.Error("Kafka GPS Location packet 0x22 error peer = %s error = %v", peer, err)
			return imei
		}
		logger.Info("============== 2G Gps Location Packet parsing end ===================")
	case ProtocolNumber(protocol.ProtoMessage):
		logger.Info("LBS multibase extended information packet 0x28 start ======================")
		data, err := lbsmultibase.DecodeLBSMultiBasePacket(frame)
		if err != nil {
			logger.Error("Error decoding in LBS multibase packet : %v", err)
			return imei
		}

		ack := lbsmultibase.EncodeLBSMultiBaseInformationPacket(data.SerialNumber, frame.ProtocolNumber())

		if _, err := conn.Write(ack); err != nil {
			logger.Error("Ack write err peer = %s error  %v", peer, err)
			return imei
		}

		request := kafkaproducer.ProducerRequest{
			Topic:    kafkaTopic,
			DeviceId: strconv.FormatInt(imei, 10),
			Packet: DevicePacket{
				ProtocolNumber: utils.HexByteUpper(byte(frame.ProtocolNumber())),
				Frame:          utils.HexUpper(frame.Raw),
				Raw:            frame.Raw,
				Ack:            utils.HexUpper(ack),
				PacketFrom:     "device",
				Time:           utils.ToEpochISTSecond(),
			},
		}

		if err := kafkaProducer.Send(request); err != nil {
			logger.Error("Kafka writing error %v", err)
		}

		logger.Info("LBS multibase extended information packet 0x28 end ======================")

	case ProtocolNumber(protocol.ProtoWifiPacket):
		logger.Info("Wifi packet 0x2C start =======================")
		request := kafkaproducer.ProducerRequest{
			Topic:    kafkaTopic,
			DeviceId: strconv.FormatInt(imei, 10),
			Packet: DevicePacket{
				ProtocolNumber: utils.HexByteUpper(byte(frame.ProtocolNumber())),
				Frame:          utils.HexUpper(frame.Raw),
				Raw:            frame.Raw,
				PacketFrom:     "device",
				Time:           utils.ToEpochISTSecond(),
			},
		}

		if err := kafkaProducer.Send(request); err != nil {
			logger.Error("Kafka Wifi writing error %v", err)
		}

		logger.Info("Wifi packet 0x2C end =======================")

	case ProtocolNumber(protocol.ProtoTimeCalibrationPacket):
		logger.Info("Time Calibration packet 0x8A start =============")

		ack := timecalibration.EncodeTimeCalibration4GAck(frame.SerialNumber(), frame.ProtocolNumber())

		if _, err := conn.Write(ack); err != nil {
			logger.Error("Write failed for Time Calibration 0x8A")
			return imei
		}
		request := kafkaproducer.ProducerRequest{
			Topic:    kafkaTopic,
			DeviceId: strconv.FormatInt(imei, 10),
			Packet: DevicePacket{
				ProtocolNumber: utils.HexByteUpper(byte(frame.ProtocolNumber())),
				Frame:          utils.HexUpper(frame.Raw),
				Raw:            frame.Raw,
				Ack:            utils.HexUpper(ack),
				PacketFrom:     "device",
				Time:           utils.ToEpochISTSecond(),
			},
		}

		if err := kafkaProducer.Send(request); err != nil {
			logger.Error("Kafka Wifi writing error %v", err)
		}
		logger.Info("Time Calibration packet 0x8A end =============")

	case ProtocolNumber(protocol.ProtoInformationTransferPacket):
		logger.Info("Information Transfer Packet 0x94 start =============")
		request := kafkaproducer.ProducerRequest{
			Topic:    kafkaTopic,
			DeviceId: strconv.FormatInt(imei, 10),
			Packet: DevicePacket{
				ProtocolNumber: utils.HexByteUpper(byte(frame.ProtocolNumber())),
				Frame:          utils.HexUpper(frame.Raw),
				Raw:            frame.Raw,
				PacketFrom:     "device",
				Time:           utils.ToEpochISTSecond(),
			},
		}

		if err := kafkaProducer.Send(request); err != nil {
			logger.Error("Kafka Wifi writing error %v", err)
		}

		logger.Info("Information Transfer Packet 0x94 end =============")

	case ProtocolNumber(protocol.ProtoGPSLocationOver4G):
		logger.Info("GPS Location over 4G 0xA0 start =============")
		request := kafkaproducer.ProducerRequest{
			Topic:    kafkaTopic,
			DeviceId: strconv.FormatInt(imei, 10),
			Packet: DevicePacket{
				ProtocolNumber: utils.HexByteUpper(byte(frame.ProtocolNumber())),
				Frame:          utils.HexUpper(frame.Raw),
				Raw:            frame.Raw,
				PacketFrom:     "device",
				Time:           utils.ToEpochISTSecond(),
			},
		}

		if err := kafkaProducer.Send(request); err != nil {
			logger.Error("Kafka Wifi writing error %v", err)
		}
		logger.Info("GPS Location over 4G 0xA0 end =============")

	case ProtocolNumber(protocol.ProtoOnlineCommandTerminal):
		logger.Info("Online command Terminal start =============")

		data, err := onlinecommand.DecodeOnlineCommandTerminalPacket(frame)
		if err != nil {
			logger.Error("Error while reading packet %v", err)
			break
		}

		packetKey := strconv.FormatInt(imei, 10) + data.ServerFlag
		rawContent := utils.DecodeStringToByte(data.Content)

		commandRequest := kafkacommandproducer.CommandProducerRequest{
			Topic:     commandTopic,
			CommandId: packetKey,
			Packet: CommandResponsePacket{
				Imei:       imei,
				ServerFlag: data.ServerFlag,
				Response:   utils.CleanDeviceText(rawContent),
				TimeStamp:  utils.ToEpochISTSecond(),
			},
		}

		if err := commandProducer.Send(commandRequest); err != nil {
			logger.Error("Kafka device command writing error %v", err)
		}
		logger.Info("Online command Terminal end =============")
	}
	return imei
}