package kafkaproducer

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

	. "primesys_producer/src/types"

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

type PacketProducer struct {
	client *kafka.Producer
}

type ProducerRequest struct {
	Topic    string
	DeviceId string
	Packet   DevicePacket
}

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

		// Reliability
		"acks":               "all",
		"enable.idempotence": true,
		"message.timeout.ms": 30000,

		// Throughput
		"compression.type": "lz4",
		"linger.ms":        5,
		"batch.size":       65536,
	})

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

	producer := &PacketProducer{
		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 *PacketProducer) deliveryReports() {
	for e := range p.client.Events() {
		switch ev := e.(type) {
		case *kafka.Message:
			if ev.TopicPartition.Error != nil {
				logger.Error("kafka push failed imei=%s protocol=%v error=%v", string(ev.Key), ev.Opaque, ev.TopicPartition.Error)
				continue
			}
			logger.Info("kafka push ok imei=%s protocol=%v topic=%s partition=%d offset=%v",
				string(ev.Key), ev.Opaque, *ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)
		case kafka.Error:
			logger.Error("kafka client error: %v", ev)
		}
	}
}

func (p *PacketProducer) Send(request ProducerRequest) 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,
			Partition: kafka.PartitionAny,
		},
		Key:           []byte(request.DeviceId),
		Value:         []byte(data),
		Timestamp:     utils.IndianTime(),
		TimestampType: kafka.TimestampCreateTime,
		Opaque:        request.Packet.ProtocolNumber,
	}, nil)
	if err != nil {
		logger.Error("kafka push not queued imei=%s protocol=%s error=%v", request.DeviceId, request.Packet.ProtocolNumber, err)
	}
	return err
}

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