package kafkaconsumer

import (
	"encoding/json"
	"primesys_producer/src/kafkacommandproducer"
	"primesys_producer/src/logger"
	onlinecommand "primesys_producer/src/protocol/online_command"
	"primesys_producer/src/store"
	"primesys_producer/src/types"
	"primesys_producer/src/utils"
	"runtime/debug"

	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
)

type CommandConsumer struct {
	client          *kafka.Consumer
	kafkaUrl        string
	groupId         string
	topic           string
	commandProducer *kafkacommandproducer.CommandProducer
}

func NewKafkaConsumer(kafkaUrl string, groupId string, topic string, commandProducer *kafkacommandproducer.CommandProducer) (*CommandConsumer, error) {
	client, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers": kafkaUrl,
		"group.id":          groupId,
		"auto.offset.reset": "latest",
	})

	if err != nil {
		logger.Warn("Error while creating kafka consumer")
		return nil, err
	}

	if err := client.SubscribeTopics([]string{topic}, nil); err != nil {
		client.Close()
		return nil, err
	}

	return &CommandConsumer{
		client:          client,
		kafkaUrl:        kafkaUrl,
		groupId:         groupId,
		topic:           topic,
		commandProducer: commandProducer,
	}, nil
}

// Consume blocks and reads messages from the subscribed topic
func (c *CommandConsumer) Consume() {
	for {
		msg, err := c.client.ReadMessage(-1)
		if err != nil {
			logger.Error("kafka consume error: %v", err)
			continue
		}
		logger.Info("kafka consume ok key=%s topic=%s partition=%d offset=%v value=%s",
			string(msg.Key), *msg.TopicPartition.Topic, msg.TopicPartition.Partition, msg.TopicPartition.Offset, string(msg.Value))

		topic := *msg.TopicPartition.Topic
		c.handleMessage(msg, topic)
	}
}

func (c *CommandConsumer) KafkaClose() {
	if c == nil {
		return
	}
	c.client.Close()
}

func (c *CommandConsumer) handleMessage(message *kafka.Message, topic string) {

	defer func() {
		if r := recover(); r != nil {
			logger.Error("panic recovered key=%s offset=%v: %v\n%s", string(message.Key), message.TopicPartition.Offset, r, debug.Stack())
		}
	}()

	var request struct {
		types.CommandRequestPacket
		Response *string `json:"response"`
	}

	err := json.Unmarshal(message.Value, &request)

	if err != nil {
		logger.Error("Json parser error : %v", err)
		return
	}

	// skip command responses produced on the same topic
	if request.Response != nil {
		return
	}

	command := request.CommandRequestPacket

	encodeTerminal, err := onlinecommand.EncodeCommandToTerminal(command.Command, command.ServerFlag)

	if err != nil {
		logger.Error("Encoding to terminal of device command failed %d : %v", command.Imei, err)
	}

	packet := onlinecommand.EncodeCommandPacket(encodeTerminal)
	logger.Info("imei=%d,command=%s,out=%s", command.Imei, command.Command, utils.HexUpper(packet))

	conn, ok := store.GetDeviceConn(command.Imei)

	if !ok {
		commandRequest := kafkacommandproducer.CommandProducerRequest{
			Topic:     topic,
			CommandId: string(message.Key),
			Packet: types.CommandResponsePacket{
				Imei:       command.Imei,
				ServerFlag: command.ServerFlag,
				Response:   "Device not connected to server",
				TimeStamp:  utils.ToEpochISTSecond(),
			},
		}
		if err := c.commandProducer.Send(commandRequest); err != nil {
			logger.Error("Kafka device command writing error %v", err)
		}
		logger.Error("Device not connected to server")
		return
	}

	if _, err := conn.Write(packet); err != nil {
		logger.Error("imei=%d command write error: %v", command.Imei, err)
	}

}