package eventbushandler

import (
	"database/sql"
	"errors"
	"sync"
	"time"

	"github.com/pingcap/dbaas/billing2/billing_errors"
	"github.com/pingcap/dbaas/billing2/config"
	"github.com/pingcap/dbaas/billing2/db"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"gorm.io/gorm"
)

type EventHandleFuc func(event *Event) error

type Event struct {
	EventID   string
	EventType string
	EventData interface{}
}

type EventBusHandler interface {
	// start run event bus handler
	Run() error

	PublishEvent(eventType string, event *Event) error

	PublishEventWithTransaction(eventType string, event *Event, tx *gorm.DB) error

	// event data should only be a struct not pointer, and can be cloned by golang := operator
	RegisterEventHandleFunc(eventType string, group string, eventDataStruct interface{}, eventFunc EventHandleFuc) error

	FixDLQ(timeout time.Duration) error

	Close() error
}

func Get() EventBusHandler {
	return global
}

var global EventBusHandler = &EventBusHandlerImpl{}

// for test only
func Set(eventBusHandler EventBusHandler) {
	global = eventBusHandler
}

type EventBusHandlerImpl struct {
	Impl           tieventbus.Eventbus
	Once           sync.Once
	handlerConfigs []tieventbus.HandlerConfig
}

func initEventbus() (tieventbus.Eventbus, error) {
	sqlDB, err := db.GetDB().DB()
	if err != nil {
		err = billing_errors.NewSystemError(err)
		return nil, err
	}
	res, err := tieventbus.NewEventbus(tieventbus.EventbusConfig{
		AppName:    "billing",
		DB:         sqlDB,
		TxPubTable: "billing_tieventbus",
	})
	if err != nil {
		err = billing_errors.NewSystemError(err)
		return nil, err
	}
	return res, err
}

func registerAllHandler(eventbus tieventbus.Eventbus, handlerConfigs []tieventbus.HandlerConfig) {
	for _, conf := range handlerConfigs {
		eventbus.RegisterHandler(conf)
	}
}

func (impl *EventBusHandlerImpl) getImpl() (tieventbus.Eventbus, error) {
	var err error = nil
	impl.Once.Do(func() {
		impl.Impl, err = initEventbus()
	})
	if err != nil {
		return nil, billing_errors.NewSystemError(err)
	}
	return impl.Impl, nil
}

func (impl *EventBusHandlerImpl) PublishEvent(eventType string, event *Event) error {
	if !config.Get().GetBool(config.EnableEventbus) {
		return billing_errors.NewSystemError(errors.New("eventbus is not enabled"))
	}
	tieventbusImpl, err := impl.getImpl()
	if err != nil {
		return err
	}
	err = tieventbusImpl.Publish(&tieventbus.Event{
		ID:   event.EventID,
		Type: eventType,
		Data: event.EventData,
	})
	if err != nil {
		return billing_errors.NewSystemError(err)
	}
	return nil
}

func (impl *EventBusHandlerImpl) PublishEventWithTransaction(eventType string, event *Event, gTx *gorm.DB) error {
	if !config.Get().GetBool(config.EnableEventbus) {
		return billing_errors.NewSystemError(errors.New("eventbus is not enabled"))
	}
	tieventbusImpl, err := impl.getImpl()
	if err != nil {
		return err
	}
	tx := gTx.Statement.ConnPool.(*sql.Tx)

	err = tieventbusImpl.PublishWithDBTx(tx, &tieventbus.Event{
		ID:   event.EventID,
		Type: eventType,
		Data: event.EventData,
	})
	if err != nil {
		return billing_errors.NewSystemError(err)
	}
	return nil
}

func (impl *EventBusHandlerImpl) Run() error {
	if !config.Get().GetBool(config.EnableEventbus) {
		return nil
	}
	tieventbusImpl, err := impl.getImpl()
	if err != nil {
		return err
	}
	registerAllHandler(tieventbusImpl, impl.handlerConfigs)
	tieventbusImpl.EnableMetrics(prometheus.DefaultRegisterer, "billing", "eventbus")
	err = tieventbusImpl.Run()
	if err != nil {
		return billing_errors.NewSystemError(err)
	}
	return nil
}

func (impl *EventBusHandlerImpl) RegisterEventHandleFunc(eventType string, group string, eventDataStruct interface{}, eventFunc EventHandleFuc) error {
	if group == "" {
		group = "default"
	}
	impl.handlerConfigs = append(impl.handlerConfigs, tieventbus.HandlerConfig{
		EventType: eventType,
		Group:     group,
		HandleFunc: func(event *tieventbus.Event) error {
			return eventFunc(&Event{
				EventID:   event.ID,
				EventType: event.Type,
				EventData: event.Data,
			})
		},
		DataStruct: eventDataStruct,
	})
	return nil
}

func (impl *EventBusHandlerImpl) FixDLQ(timeout time.Duration) error {
	eb, err := initEventbus()
	if err != nil {
		return err
	}
	registerAllHandler(eb, impl.handlerConfigs)
	err = eb.FixDLQ(timeout)
	if err != nil {
		return billing_errors.NewSystemError(err)
	}
	return nil
}

func (impl *EventBusHandlerImpl) Close() error {
	if impl.Impl != nil { //nolint:typecheck
		impl.Impl.Close()
	}

	return nil
}