package cost

import (
	"context"
	"fmt"
	"net/url"
	"sync"

	"github.com/Shopify/sarama"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/cost-collect/pkg/config"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/zap"
	"golang.org/x/net/proxy"
)

var (
	eventbusOnce sync.Once
	gEventbus    tieventbus.Eventbus
)

func NewEventbus(eventbusConf config.Eventbus) tieventbus.Eventbus {
	eventbusOnce.Do(func() {
		var saramaConfigModifier tieventbus.SaramaConfigModifier = func(conf *sarama.Config) {
			if eventbusConf.ProxyAddress != "" {
				log.Info(context.Background(), "use eventbus proxy", zap.String("address", eventbusConf.ProxyAddress))
				// enable proxy to support access tieventbus in your laptop(or local develop environment)
				proxyURL, err := url.Parse(eventbusConf.ProxyAddress)
				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
			}

			// pull messages from the newest point
			conf.Consumer.Offsets.Initial = sarama.OffsetNewest
		}
		eventbus, err := tieventbus.NewEventbus(
			tieventbus.EventbusConfig{
				AppName:                         EventBusAppName,
				SaramaConfigModifier4Producer:   saramaConfigModifier,
				SaramaConfigModifier4Subscriber: saramaConfigModifier,
			})
		if err != nil {
			panic(fmt.Sprintf("failed to initialize event bus, err=%s", err.Error()))
		}
		eventbus.EnableMetrics(prometheus.DefaultRegisterer, EventBusAppName, "eventbus")

		gEventbus = eventbus
	})

	return gEventbus
}