package main

import (
	"flag"
	"fmt"
	"primesys_producer/src/config"
	"primesys_producer/src/kafkacommandproducer"
	"primesys_producer/src/kafkaconsumer"
	"primesys_producer/src/kafkaproducer"
	"primesys_producer/src/logger"
	"primesys_producer/src/redis"
	"primesys_producer/src/server"
	"primesys_producer/src/websocket"
	"strconv"

	"github.com/joho/godotenv"
)

var addr = flag.String("addr", ":5656", "Producer Server address")

func main() {
	fmt.Println("Welcome to primesys producer")
	fmt.Println("Producer address", *addr)
	flag.Parse()

	// dot env parsing
	err := godotenv.Load()

	if err != nil {
		logger.Fatal("Env load error")
	}

	cfg := config.LoadEnv()

	if err := redis.ConnectRedis(cfg.RedisAddr, cfg.RedisPassword); err != nil {
		logger.Fatal("Redis intialization failed: %v", err)
	}

	defer redis.CloseRedis()

	logger.Info("Redis Connected.....")

	var kafkaProducer *kafkaproducer.PacketProducer
	var kafkaCommandProducer *kafkacommandproducer.CommandProducer
	var kafkaConsumer *kafkaconsumer.CommandConsumer

	kafkaEnable, err := strconv.ParseBool(cfg.KafkaEnabled)

	if err != nil {
		logger.Fatal("Kafka Error in starting.....")
	}

	if kafkaEnable {
		kafkaProducer, err = kafkaproducer.NewKafkaProducer(cfg.KafkaBrokers)

		if err != nil {
			logger.Fatal("Kafka producer initialization failed : %v", err)
		}

		defer kafkaProducer.KafkaClose()

		logger.Info("Kafka Producer started...")

		kafkaCommandProducer, err = kafkacommandproducer.NewKafkaCommandProducer(cfg.KafkaConsumerBrokers)

		if err != nil {
			logger.Fatal("Kafka command producer initialization failed : %v", err)
		}

		defer kafkaCommandProducer.KafkaClose()

		logger.Info("Kafka Command Producer started...")

		kafkaConsumer, err = kafkaconsumer.NewKafkaConsumer(cfg.KafkaConsumerBrokers, cfg.KafkaConsumerGroup, cfg.DeviceCommand, kafkaCommandProducer)

		if err != nil {
			logger.Fatal("Kafka consumer initialization failed : %v", err)
		}

		defer kafkaConsumer.KafkaClose()

		go kafkaConsumer.Consume()

		logger.Info("Kafka Consumer started...")
	} else {
		logger.Warn("Kafka disabled (KAFKA_ENABLED=false)")
	}

	go websocket.SocketServer(cfg.SocketPort)

	server.TCPServer(cfg.AppPort, cfg.KafkaTopic, cfg.DeviceCommand, kafkaProducer, kafkaCommandProducer, kafkaConsumer)
}