package kafkacommandproducer

import (
	"primesys_producer/src/logger"
	"primesys_producer/src/utils"

	. "primesys_producer/src/types"

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

type CommandProducer struct {
	client *kafka.Producer
}

type CommandProducerRequest struct {
	Topic     string
	CommandId string
	Packet    CommandResponsePacket
}

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

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

	producer := &CommandProducer{
		client: client,
	}
	go producer.deliveryReports()

	return producer, nil
}

// Produce only queues a message; the broker's result arrives here.
// This channel must be drained, otherwise it fills up and producing stalls.
func (p *CommandProducer) deliveryReports() {
	for e := range p.client.Events() {
		switch ev := e.(type) {
		case *kafka.Message:
			if ev.TopicPartition.Error != nil {
				logger.Error("kafka command push failed imei=%s error=%v", string(ev.Key), ev.TopicPartition.Error)
				continue
			}
			logger.Info("kafka command push ok imei=%s topic=%s partition=%d offset=%v",
				string(ev.Key), *ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)
		case kafka.Error:
			logger.Error("kafka command client error: %v", ev)
		}
	}
}

func (p *CommandProducer) Send(request CommandProducerRequest) error {
	if p == nil {
		return nil
	}

	data, err := utils.StructToJsonString(request.Packet)

	if err != nil {
		return err
	}

	err = p.client.Produce(&kafka.Message{
		TopicPartition: kafka.TopicPartition{
			Topic: &request.Topic,
		},
		Key:           []byte(request.CommandId),
		Value:         []byte(data),
		Timestamp:     utils.IndianTime(),
		TimestampType: kafka.TimestampCreateTime,
	}, nil)
	if err != nil {
		logger.Error("kafka command push not queued imei=%d error=%v", request.Packet.Imei, err)
	}
	return err
}

func (p *CommandProducer) KafkaClose() {
	if p == nil {
		return
	}
	p.client.Flush(5000)
	p.client.Close()
}