package main

import (
	"context"
	"flag"
	"fmt"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"github.com/tidbcloud/lake-ingest-kafka/config"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/sirupsen/logrus"
)

var (
	kafkaBrokers       = []string{"localhost:9092"}
	kafkaTopic         = "test"
	kafkaConsumerGroup = "test-group"
	batchSize          = 1000
	batchMaxInterval   = 5 * time.Second
	dataFormat         = "json"
	lakeDSN        = "root:@tcp(127.0.0.1:3306)/test"
	lakeTable      = "default.test"
)

var version = "dev"

func main() {
	configFile := flag.String("f", "./config/conf.json", "Path to the configuration file")
	showVersion := flag.Bool("v", false, "Print bend-ingest-kafka version and exit")
	flag.BoolVar(showVersion, "version", false, "Print bend-ingest-kafka version and exit")
	flag.Parse()

	if *showVersion {
		fmt.Printf("bend-ingest-kafka version %s\n", version)
		return
	}

	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		sigch := make(chan os.Signal, 1)
		signal.Notify(sigch, syscall.SIGQUIT, syscall.SIGTERM)
		<-sigch
		cancel()
	}()

	cfg := parseConfig(configFile)
	ig := NewLakeIngester(cfg)
	defer ig.Close()

	initMetrics()
	go func() {
		mux := http.NewServeMux()
		mux.Handle("/metrics", promhttp.Handler())
		addr := fmt.Sprintf(":%d", cfg.MetricsPort)
		logrus.Infof("Prometheus metrics available at %s/metrics", addr)
		if err := http.ListenAndServe(addr, mux); err != nil {
			logrus.Errorf("metrics server error: %v", err)
		}
	}()

	if !cfg.IsJsonTransform {
		err := ig.CreateRawTargetTable()
		if err != nil {
			panic(err)
		}
	}

	wg := sync.WaitGroup{}
	wg.Add(cfg.Workers)
	for i := 0; i < cfg.Workers; i++ {
		w := NewConsumeWorker(cfg, fmt.Sprintf("worker-%d", i), ig)
		go func() {
			w.Run(ctx)
			wg.Done()
		}()
	}
	wg.Wait()
}

func parseConfigWithFile(configFile *string) *config.Config {
	cfg, err := config.LoadConfig(configFile)
	if err != nil {
		panic(err)
	}
	validateConfig(cfg)
	return cfg
}

func validateConfig(cfg *config.Config) {
	if cfg.IsJsonTransform && cfg.UseReplaceMode {
		panic("replace mode can only be used when is-json-transform is false")
	}
	if cfg.UseStreamingLoad && cfg.IsJsonTransform {
		panic("streaming load mode can only be used when is-json-transform is false (raw mode)")
	}
	if cfg.UseStreamingLoad && cfg.UseReplaceMode {
		panic("streaming load mode cannot be combined with replace mode")
	}
	if cfg.UseStreamingLoad && cfg.IsJsonTransform {
		panic("streaming load mode can only be used when is-json-transform is false (raw mode)")
	}
	if cfg.UseStreamingLoad && cfg.UseReplaceMode {
		panic("streaming load mode cannot be combined with replace mode")
	}
	if cfg.IsSASL && cfg.SaslUser == "" {
		panic("sasl user is required when sasl is enabled")
	}
	if cfg.DisableTLS && !cfg.IsSASL {
		logrus.Warn("disableTLS is enabled without SASL. Connection will use PLAINTEXT protocol.")
	}
	if cfg.DisableTLS && cfg.IsSASL {
		logrus.Warn("disableTLS is enabled with SASL. Credentials will be transmitted without encryption (SASL_PLAINTEXT). Only use in trusted networks.")
	}
}

func parseConfig(configFile *string) *config.Config {
	if *configFile == "" {
		*configFile = "config/conf.json"
	}

	// Detect whether the user explicitly passed -f on the command line.
	fileFlagSet := false
	flag.Visit(func(fl *flag.Flag) {
		if fl.Name == "f" {
			fileFlagSet = true
		}
	})

	if _, err := os.Stat(*configFile); err == nil {
		return parseConfigWithFile(configFile)
	} else if fileFlagSet {
		// User explicitly specified a config file but it cannot be used.
		panic(fmt.Sprintf("config file %q not found or not accessible: %v", *configFile, err))
	}

	cfg := config.Config{}
	flag.StringVar(&cfg.KafkaBootstrapServers, "kafka-bootstrap-servers", "127.0.0.1:64103", "Kafka bootstrap servers")
	flag.StringVar(&cfg.KafkaTopic, "kafka-topic", "test", "Kafka topic")
	flag.StringVar(&cfg.KafkaConsumerGroup, "kafka-consumer-group", "kafka-bend-ingest", "Kafkaconsumer group")
	flag.StringVar(&cfg.LakeDSN, "lake-dsn", "http://root:root@localhost:8002", "Lake DSN")
	flag.StringVar(&cfg.LakeTable, "lake-table", "test_ingest", "Lake table")
	flag.StringVar(&cfg.MockData, "mock-data", "", "generate mock data to lake")
	flag.IntVar(&cfg.BatchSize, "batch-size", 1024, "Batch size")
	flag.IntVar(&cfg.Workers, "workers", 1, "Number of workers")
	flag.IntVar(&cfg.BatchMaxInterval, "batch-max-interval", 30, "Batch max interval")
	flag.StringVar(&cfg.DataFormat, "data-format", "json", "kafka data format")
	flag.BoolVar(&cfg.CopyPurge, "copy-purge", false, "purge data before copy")
	flag.BoolVar(&cfg.CopyForce, "copy-force", false, "force copy data")
	flag.BoolVar(&cfg.IsJsonTransform, "is-json-transform", true, "transform json data")
	flag.BoolVar(&cfg.DisableVariantCheck, "disable-variant-check", false, "disable variant check")
	flag.IntVar(&cfg.MinBytes, "min-bytes", 1024, "min bytes")
	flag.IntVar(&cfg.MaxBytes, "max-bytes", 20*1024*1024, "max bytes")
	flag.IntVar(&cfg.MaxWait, "max-wait", 10, "max wait")
	flag.BoolVar(&cfg.UseReplaceMode, "use-replace-mode", false, "use replace into mode")
	flag.StringVar(&cfg.UserStage, "user-stage", "~", "user stage")
	flag.BoolVar(&cfg.UseStreamingLoad, "use-streaming-load", false, "use /v1/streaming_load HTTP endpoint (raw mode only)")
	flag.BoolVar(&cfg.CopyIntoUploadCompression, "copy-into-upload-compression", true, "enable zstd compression for staged NDJSON files used by COPY INTO")
	flag.IntVar(&cfg.MetricsPort, "metrics-port", 2112, "Prometheus metrics HTTP port")
	flag.BoolVar(&cfg.EnableRebalanceOptimization, "enable-rebalance-optimization", true, "enable faster partition-change detection, explicit assignment strategy, and rebalance logging")
	flag.StringVar(&cfg.PartitionAssignmentStrategy, "partition-assignment-strategy", "range,roundrobin", "partition assignment strategy list, for example: range,roundrobin")
	flag.IntVar(&cfg.TopicMetadataRefreshIntervalMs, "topic-metadata-refresh-interval-ms", 60000, "topic metadata refresh interval in ms; lower values detect added partitions faster")

	flag.Parse()
	validateConfig(&cfg)
	return &cfg
}