package consumer

import (
	"net/url"
	"time"

	"github.com/Shopify/sarama"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/metrics"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"golang.org/x/net/proxy"
)

const (
	appName = "dataflow-service"
	group   = "default-consumer"
)

func RegisterConsumer() {
	eventbus := buildEventBus(false)
	if eventbus == nil {
		return
	}
	eventbus.EnableMetrics(prometheus.DefaultRegisterer, metrics.DataflowNamespace, "eventbus")
	err := eventbus.Run()
	if err != nil {
		panic(err)
	}
}

func RegisterDLQConsumer(timeout time.Duration) error {
	eventbus := buildEventBus(true)
	return eventbus.FixDLQ(timeout)
}

func buildEventBus(isDLQ bool) tieventbus.Eventbus {
	if config.Global().DebugConfig.Enable && config.Global().DebugConfig.DisableKafka {
		return nil
	}
	var saramaConfigModifier tieventbus.SaramaConfigModifier = func(conf *sarama.Config) {
		if config.Global().DebugConfig.Enable {
			// enable proxy to support access tieventbus in your laptop(or local develop environment)
			proxyURL, err := url.Parse(config.Global().DebugConfig.KafkaProxyAddress)
			if err != nil {
				panic(err)
			}
			p, err := proxy.FromURL(proxyURL, proxy.Direct)
			if err != nil {
				panic(err)
			}
			conf.Net.Proxy.Enable = true
			conf.Net.Proxy.Dialer = p
		}
		if !isDLQ {
			// pull messages from newest point
			conf.Consumer.Offsets.Initial = sarama.OffsetNewest
		} else {
			// for dlq consumer, pull messages from oldest point
			conf.Consumer.Offsets.Initial = sarama.OffsetOldest
		}
	}
	eventbus, err := tieventbus.NewEventbus(tieventbus.EventbusConfig{
		AppName:                         appName,
		MQTimeout:                       time.Second * 2,
		SaramaConfigModifier4Subscriber: saramaConfigModifier,
		SaramaConfigModifier4Producer:   saramaConfigModifier,
	})
	if err != nil {
		panic(err)
	}
	eventbus.RegisterHandler(
		buildCMEventConsumer(),
	)
	return eventbus
}