package tieventbus

import (
	"database/sql"
	"sync"

	"github.com/Shopify/sarama"
	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill-kafka/v2/pkg/kafka"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/ThreeDotsLabs/watermill/pubsub/gochannel"
	"github.com/spf13/viper"
	ov "github.com/tidbcloud/config-service/oneconfig/viper"
	"github.com/tidbcloud/tiinf/transport/tieventbus/tidb"
)

const KafkaBrokersKey = "kafka_brokers"

type pubSub interface {
	GetMQPublisher() (message.Publisher, error)
	GetMQSubscriber(group string) (message.Subscriber, error)
}

type pubSubImpl struct {
	eventbusConf  *EventbusConfig
	viper         *viper.Viper
	publisher     message.Publisher
	subLock       *sync.Mutex
	subscriberMap map[string]message.Subscriber
	logger        watermill.LoggerAdapter
}

func (ps *pubSubImpl) initKafkaPublisher() (err error) {
	saramaConfig := kafka.DefaultSaramaSyncPublisherConfig()
	saramaConfig.Admin.Timeout = ps.eventbusConf.MQTimeout
	saramaConfig.Metadata.Timeout = ps.eventbusConf.MQTimeout
	saramaConfig.Producer.Timeout = ps.eventbusConf.MQTimeout
	if ps.eventbusConf.SaramaConfigModifier4Producer != nil {
		logger.Info("SaramaConfigModifier4Producer configured", nil)
		ps.eventbusConf.SaramaConfigModifier4Producer(saramaConfig)
	}
	ps.publisher, err = kafka.NewPublisher(
		kafka.PublisherConfig{
			Brokers:               ps.getKafkaBrokers(),
			Marshaler:             kafka.DefaultMarshaler{},
			OverwriteSaramaConfig: saramaConfig,
		},
		ps.logger,
	)
	return
}

func (ps *pubSubImpl) GetMQPublisher() (message.Publisher, error) {
	return ps.publisher, nil
}

func (ps *pubSubImpl) GetMQSubscriber(group string) (message.Subscriber, error) {
	if ps.subscriberMap[group] != nil {
		return ps.subscriberMap[group], nil
	}
	ps.subLock.Lock()
	defer ps.subLock.Unlock()
	if ps.subscriberMap[group] == nil {
		saramaConfig := kafka.DefaultSaramaSubscriberConfig()
		saramaConfig.Consumer.Offsets.Initial = sarama.OffsetOldest
		saramaConfig.Admin.Timeout = ps.eventbusConf.MQTimeout
		saramaConfig.Metadata.Timeout = ps.eventbusConf.MQTimeout
		if ps.eventbusConf.SaramaConfigModifier4Subscriber != nil {
			logger.Info("SaramaConfigModifier4Subscriber configured", nil)
			ps.eventbusConf.SaramaConfigModifier4Subscriber(saramaConfig)
		}
		sub, err := kafka.NewSubscriber(
			kafka.SubscriberConfig{
				Brokers:               ps.getKafkaBrokers(),
				Unmarshaler:           kafka.DefaultMarshaler{},
				ConsumerGroup:         group,
				OverwriteSaramaConfig: saramaConfig,
			},
			ps.logger,
		)
		if err != nil {
			return nil, err
		}
		ps.subscriberMap[group] = sub
	}
	return ps.subscriberMap[group], nil
}

func (ps *pubSubImpl) initKafkaConfig() error {
	ps.viper = viper.New()
	ps.viper.SetDefault(KafkaBrokersKey, []string{"localhost:9092"})
	remote := "oneconfig-eventbus/kafka-config.yaml"
	logger.Info("Read From Remote", watermill.LogFields{"remote": remote})
	ps.viper.SetConfigType("yaml")
	err := ps.viper.AddRemoteProvider(ov.Provider, ov.Endpoint, remote)
	if err != nil {
		return err
	}
	err = ps.viper.WatchRemoteConfig()
	if err != nil {
		if err != viper.RemoteConfigError("No Files Found") {
			return err
		}
	}
	return nil
}

func (ps *pubSubImpl) getKafkaBrokers() []string {
	brokers := ps.viper.GetStringSlice(KafkaBrokersKey)
	return brokers
}

func newPubSubImpl(conf EventbusConfig) (*pubSubImpl, error) {
	ps := &pubSubImpl{
		logger:        logger,
		subscriberMap: make(map[string]message.Subscriber),
		subLock:       &sync.Mutex{},
		eventbusConf:  &conf,
	}
	err := ps.initKafkaConfig()
	if err != nil {
		return nil, err
	}
	err = ps.initKafkaPublisher()
	if err != nil {
		return nil, err
	}
	return ps, nil
}

type pubSubMock struct {
	channel *gochannel.GoChannel
	logger  watermill.LoggerAdapter
}

func (ps *pubSubMock) GetMQPublisher() (message.Publisher, error) {
	return ps.channel, nil
}

func (ps *pubSubMock) GetMQSubscriber(group string) (message.Subscriber, error) {
	return ps.channel, nil
}

func newPubSubMock() (*pubSubMock, error) {
	channel := gochannel.NewGoChannel(gochannel.Config{
		Persistent: true,
	}, logger)
	return &pubSubMock{
		channel: channel,
		logger:  logger,
	}, nil
}

func getSQLSubscriber(db *sql.DB) (*tidb.Subscriber, error) {
	return tidb.NewSubscriber(
		db,
		tidb.SubscriberConfig{
			InitializeSchema: true,
		},
		logger,
	)
}

func getSQLTxPublisher(tx *sql.Tx) (*tidb.Publisher, error) {
	return tidb.NewPublisher(
		tx,
		tidb.PublisherConfig{},
		logger,
	)
}

type disablePubSub struct {
}

func (ps *disablePubSub) GetMQPublisher() (message.Publisher, error) {
	return &disabledPublisher{}, nil
}

func (ps *disablePubSub) GetMQSubscriber(group string) (message.Subscriber, error) {
	return nil, nil
}