package publisher

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

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

type eventBusPublisherImpl struct {
	eb tieventbus.Eventbus
}

func NewEventBusPublisherImpl(appName string) Publisher {
	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
		}
		// 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 {
		log.L().Panic(fmt.Sprintf("failed to initialize event bus, err=%s", err.Error()))
	}

	eb.EnableMetrics(prometheus.DefaultRegisterer, appName, "eventbus")
	if err := eb.Run(); err != nil {
		log.L().Panic(fmt.Sprintf("failed to run event bus, err=%s", err.Error()))
	}

	return &eventBusPublisherImpl{
		eb: eb,
	}
}

func (p *eventBusPublisherImpl) Publish(ctx context.Context, topic event.Topic, meta *event.Meta, event interface{}) error {
	if checkEventIsPtr(event) {
		return fmt.Errorf("event is not a ptr, is %s", reflect.ValueOf(event).Kind().String())
	}

	return errors.Annotate(p.eb.Publish(&tieventbus.Event{
		ID:   meta.EventID,
		Type: string(topic),
		Data: event,
	}), "eventbus publish event failed")
}

func checkEventIsPtr(event interface{}) bool {
	return reflect.ValueOf(event).Kind() != reflect.Ptr
}