package kafka

import (
	"context"
	"encoding/json"
	"fmt"
	"primesys_command_producer/src/data"
	"primesys_command_producer/src/logger"
	"primesys_command_producer/src/models"
	"primesys_command_producer/src/repositories"
	"primesys_command_producer/src/storage"
	"primesys_command_producer/src/types"
	"primesys_command_producer/src/utils"
	"primesys_command_producer/src/websocket"
	"runtime/debug"
	"strconv"
	"time"

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

type DeviceCommandResponsePacket struct {
	Imei       int64  `json:"imei"`
	ServerFlag string `json:"server_flag"`
	Response   string `json:"response"`
	TimeStamp  int64  `json:"timestamp"`
}

type CommandConsumer struct {
	client  *ckafka.Consumer
	groupId string
	topic   string
}

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

	if err != nil {
		logger.Error("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,
		groupId: groupId,
		topic:   topic,
	}, nil
}

// Consume blocks and reads messages from the subscribed topic continuously
func (c *CommandConsumer) Consume(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
			// short timeout so ctx is checked even when no messages arrive
			msg, err := c.client.ReadMessage(100 * time.Millisecond)
			if err != nil {
				if kErr, ok := err.(ckafka.Error); ok && kErr.Code() == ckafka.ErrTimedOut {
					continue
				}
				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))

			c.handleMessage(msg)
		}
	}
}

func (c *CommandConsumer) handleMessage(message *ckafka.Message) {

	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 packet struct {
		DeviceCommandResponsePacket
		Response *string `json:"response"`
	}

	if err := json.Unmarshal(message.Value, &packet); err != nil {
		logger.Error("Json parser error : %v", err)
		return
	}

	// skip command requests produced on the same topic, only device responses are handled
	if packet.Response == nil {
		return
	}

	response := packet.DeviceCommandResponsePacket
	response.Response = *packet.Response

	// responses to commands not sent by this system (old platform / sms) are ignored
	if !data.IsOurServerFlag(response.ServerFlag) {
		logger.Debug("ignore response with foreign server_flag=%s imei=%d", response.ServerFlag, response.Imei)
		return
	}

	logger.Info("Device response received imei=%d server_flag=%s response=%s", response.Imei, response.ServerFlag, response.Response)

	// search key in command.json is imei + server flag
	searchKey := strconv.FormatInt(response.Imei, 10) + response.ServerFlag

	cmd, found, err := storage.FindCommandByServerFlag(searchKey)
	if err != nil {
		logger.Error("Error searching command.json for server_flag %s: %v", searchKey, err)
		return
	}
	if !found {
		logger.Error("No command found in command.json for server_flag %s", searchKey)
		return
	}

	logger.Info("Command found in command.json for server_flag %s: %+v", searchKey, cmd)

	if err := insertIntoCommandHistory(response, cmd); err != nil {
		logger.Error("Error inserting command history for server_flag %s: %v", searchKey, err)
		return
	}

	assigned, err := storage.DeleteCommandAndAssignFlag(searchKey)
	if err != nil {
		logger.Error("Error deleting command for server_flag %s: %v", searchKey, err)
		return
	}

	logger.Info("Command history saved and removed from command.json for server_flag %s", searchKey)

	if assigned != nil {
		logger.Info("Server flag %s assigned to imei=%d userId=%s", assigned.ServerFlag, assigned.Command.Imei, assigned.UserId)

		// send only the command that just got the server flag
		if err := websocket.SendCommand(*assigned); err != nil {
			logger.Error("Error sending command to imei %d: %v", assigned.Command.Imei, err)
		}
	}
}

func insertIntoCommandHistory(response DeviceCommandResponsePacket, queueCmd types.QueueCommand) error {

	device, err := repositories.GetDeviceByImei(context.Background(), queueCmd.Command.Imei)
	if err != nil {
		return err
	}
	if device == nil {
		return fmt.Errorf("device not found for imei %d", queueCmd.Command.Imei)
	}

	division, err := repositories.GetDivisionByID(context.Background(), device.DivisionID)
	if err != nil {
		return err
	}
	userLogin, err := repositories.GetDivisionByID(context.Background(), queueCmd.UserId)
	if err != nil {
		return err
	}

	if division == nil {
		return fmt.Errorf("division not found for id %s", device.DivisionID)
	}

	command := models.CommandHistory{
		DeviceImei:         queueCmd.Command.Imei,
		Timestamp:          queueCmd.Command.TimeStamp,
		DeviceName:         device.DeviceName,
		LoginName:          userLogin.Name,
		DivisionID:         device.DivisionID,
		OriginalCommand:    queueCmd.Command.Command,
		Command:            queueCmd.Command.Command,
		DeliveredMessage:   queueCmd.DevliveredMessage,
		DeviceResponse:     response.Response,
		DeviceResponseTime: utils.IndianEpoch(),
	}

	return repositories.InsertCommandHistory(context.Background(), &command)
}

func (c *CommandConsumer) KafkaClose() {
	c.client.Close()
}