package clusterevent

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

	"github.com/Shopify/sarama"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/zap"
	"golang.org/x/net/proxy"

	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

var (
	eventbusOnce sync.Once
	gEventbus    tieventbus.Eventbus
)

//go:generate go run go.uber.org/mock/mockgen@latest -source=$GOFILE -package=mock_$GOPACKAGE -destination=./mock_$GOPACKAGE/mock_$GOFILE
type Publisher interface {
	Publish(ctx context.Context, meta *Meta, ev Event) bizerr.BizError
}

type tiEventBusPublisher struct {
	eventBus tieventbus.Eventbus
}

func NewPublisher(eventBus tieventbus.Eventbus) Publisher {
	return &tiEventBusPublisher{
		eventBus: eventBus,
	}
}

func (h *tiEventBusPublisher) Publish(ctx context.Context, meta *Meta, ev Event) bizerr.BizError {
	outEvent := &tieventbus.Event{
		ID:   meta.UUID,
		Type: ev.GetTargetTopic().String(),
		Data: ev,
	}
	tilog.Info(
		ctx,
		"try to publish event to event bus",
		zap.String("uuid", meta.UUID),
		zap.String("topic", outEvent.Type),
	)
	if err := h.eventBus.Publish(outEvent); err != nil {
		return bizerr.WrapErrorWithCode(err, bizerr.ErrCodeMQ, "publish to eventbus")
	}
	return nil
}

func NewEventbus(appName, proxyURL string) tieventbus.Eventbus {
	eventbusOnce.Do(func() {
		var saramaConfigModifier tieventbus.SaramaConfigModifier = func(conf *sarama.Config) {
			if proxyURL != "" {
				// enable proxy to support access tieventbus in your laptop(or local develop environment)
				proxyURL, err := url.Parse(proxyURL)
				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:                         appName,
				SaramaConfigModifier4Producer:   saramaConfigModifier,
				SaramaConfigModifier4Subscriber: saramaConfigModifier,
			})
		if err != nil {
			panic(fmt.Sprintf("failed to initialize event bus, err=%s", err.Error()))
		}
		eventbus.EnableMetrics(prometheus.DefaultRegisterer, "", "")

		gEventbus = eventbus
	})

	return gEventbus
}