package center

import (
	"context"
	"net/url"

	"github.com/Shopify/sarama"
	"github.com/google/uuid"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/dataflow-service-ng/internal/config"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/eventbus-datastruct/event"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/zap"
	"golang.org/x/net/proxy"
)

const (
	dataflowNamespace = "dataflow"
	appName           = "dataflow_service_status_change"
)

type EventCenterPublisher interface {
	PublishAsync(data *event.DisplayEvent)
}

type eventCenterPublisherImpl struct {
	eb tieventbus.Eventbus
}

func NewEventCenterPublisher() (EventCenterPublisher, error) {
	logger := log.GetLogger(context.Background())
	if config.Global().DebugConfig.Enable && config.Global().DebugConfig.DisableKafka {
		logger.Warn("kafka is disabled, use mock publisher")
		return &eventCenterPublisherImpl{}, 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
		}
		// pull messages from newest point
		conf.Consumer.Offsets.Initial = sarama.OffsetNewest
	}
	eb, err := tieventbus.NewEventbus(tieventbus.EventbusConfig{
		AppName:                         appName,
		SaramaConfigModifier4Producer:   saramaConfigModifier,
		SaramaConfigModifier4Subscriber: saramaConfigModifier,
	})
	if err != nil {
		logger.Error("failed to initialize event bus", zap.Error(err))
		return nil, err
	}
	eb.EnableMetrics(prometheus.DefaultRegisterer, dataflowNamespace, "eventbus")
	err = eb.Run()
	if err != nil {
		logger.Error("failed to run event bus", zap.Error(err))
		return nil, err
	}
	logger.Info("event center publisher init finished!")
	return &eventCenterPublisherImpl{
		eb: eb,
	}, nil
}

func (ec *eventCenterPublisherImpl) PublishAsync(data *event.DisplayEvent) {
	if data == nil {
		return
	}

	logger := log.GetLogger(context.Background())
	if config.Global().DebugConfig.Enable && config.Global().DebugConfig.DisableKafka {
		logger.Warn("kafka is disabled, use mock publisher", zap.Any("data", data))
		return
	}
	ec.eb.PublishAsync(&tieventbus.Event{
		ID:   uuid.NewString(),
		Type: event.DisplayEventTopic,
		Data: data,
	})

	logger.Info("event publish dispatcher: sending message finished!",
		zap.String("displayEventType", data.DisplayEventType),
		zap.String("displayStatus", string(data.DisplayStatus)),
		zap.String("contextID", data.ContextID),
		zap.String("summary", data.Summary))
}