package server

import (
	"context"
	"fmt"
	"primesys_command_producer/src/logger"
	"primesys_command_producer/src/models"
	"primesys_command_producer/src/repositories"
	. "primesys_command_producer/src/types"
	"primesys_command_producer/src/utils"
	"strconv"

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

type PacketProducer struct {
	client *kafka.Producer
}

type ProducerRequest struct {
	Topic  string
	UserId string
	Packet CommandRequestPacket
}

func NewKafkaProducerServer(kafkaUrl string) (*PacketProducer, error) {
	client, err := kafka.NewProducer(&kafka.ConfigMap{
		"bootstrap.servers": kafkaUrl,
	})

	if err != nil {
		logger.Error("Error while producing kafka channel")
		return nil, err
	}

	producer := &PacketProducer{
		client: client,
	}

	go producer.deliverReports()

	return producer, nil
}

func (p *PacketProducer) deliverReports() {
	for e := range p.client.Events() {
		switch ev := e.(type) {
		case *kafka.Message:
			if ev.TopicPartition.Error != nil {
				logger.Error("Kafka Push failed for imei=%s", string(ev.Key))
				continue
			}
			logger.Info("Kafka push ok imei=%s", string(ev.Key))
		case kafka.Error:
			logger.Error("Kafka client error: %v", ev)
		}
	}
}

func (p *PacketProducer) Send(request ProducerRequest) error {
	data, err := utils.StructToJsonString(request.Packet)

	if err != nil {
		return err
	}

	packetKey := strconv.FormatInt(request.Packet.Imei, 10) + request.Packet.ServerFlag

	err = p.client.Produce(&kafka.Message{
		TopicPartition: kafka.TopicPartition{
			Topic: &request.Topic,
		},
		Key:           []byte(packetKey),
		Value:         []byte(data),
		Timestamp:     utils.IndianTime(),
		TimestampType: kafka.TimestampCreateTime,
		Opaque:        request.Packet.Imei,
	}, nil)

	if err != nil {
		logger.Error("Kafka push not queued imei=%d userid=%s error=%v", request.Packet.Imei, request.Packet.UserID, err)
	}

	packet := request.Packet
	device, err := repositories.GetDeviceByImei(context.Background(), packet.Imei)
	if err != nil || device == nil {
		logger.Error("Device not found imei=%d err=%v", packet.Imei, err)
		return fmt.Errorf("device not found imei=%d", packet.Imei)
	}

	userLogin, err := repositories.GetDivisionByID(context.Background(), packet.UserID)
	if err != nil || userLogin == nil {
		logger.Error("No division got by this id=%s err=%v", packet.UserID, err)
		return fmt.Errorf("division not found id=%s", packet.UserID)
	}

	command := models.CommandHistory{
		DeviceImei:      packet.Imei,
		Timestamp:       utils.IndianEpoch(),
		DeviceName:      device.DeviceName + strconv.Itoa(device.DeviceNo),
		LoginName:       userLogin.Name,
		DivisionID:      device.DivisionID,
		OriginalCommand: packet.Command,
		Command:         packet.Command,
	}

	err = repositories.InsertCommandHistory(context.Background(), &command)

	if err != nil {
		logger.Error("Command Insertion Error")
	}

	return err
}

func (p *PacketProducer) KafkaSocketClose() {
	p.client.Flush(500)
	p.client.Close()
}