package tests

import (
	"database/sql"
	"time"

	"github.com/pingcap/dbaas/billing2/billing_errors"
	"github.com/pingcap/dbaas/billing2/db"
	"github.com/pingcap/dbaas/billing2/handler/eventbushandler"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"gorm.io/gorm"
)

type LocalEventBusHandlerImpl struct {
	Events         []*eventbushandler.Event
	TxEvents       []*eventbushandler.Event
	mockEventbus   tieventbus.Eventbus
	handlerConfigs []tieventbus.HandlerConfig
}

func (impl *LocalEventBusHandlerImpl) PublishEvent(eventType string, event *eventbushandler.Event) error {
	event.EventType = eventType
	tieventbusImpl := impl.mockEventbus
	err := tieventbusImpl.Publish(&tieventbus.Event{
		ID:   event.EventID,
		Type: eventType,
		Data: event.EventData,
	})
	if err != nil {
		return err
	}
	impl.Events = append(impl.Events, event)
	return nil
}

func (impl *LocalEventBusHandlerImpl) PublishEventWithTransaction(eventType string, event *eventbushandler.Event, gTx *gorm.DB) error {
	event.EventType = eventType
	tieventbusImpl := impl.mockEventbus
	tx := gTx.Statement.ConnPool.(*sql.Tx)

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

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

func (impl *LocalEventBusHandlerImpl) filterEvents(events []*eventbushandler.Event, tp ...string) []*eventbushandler.Event {
	var ret []*eventbushandler.Event
	for _, e := range events {
		if len(tp) == 0 {
			ret = append(ret, e)
		} else {
			for _, t := range tp {
				if e.EventType == t {
					ret = append(ret, e)
				}
			}
		}
	}
	return ret
}

func (impl *LocalEventBusHandlerImpl) GetAllEvents(tp ...string) []*eventbushandler.Event {
	return impl.filterEvents(impl.Events, tp...)
}

func (impl *LocalEventBusHandlerImpl) GetAllTxEvents(tp ...string) []*eventbushandler.Event {
	return impl.filterEvents(impl.TxEvents, tp...)
}

func (impl *LocalEventBusHandlerImpl) CleanEvents() {
	impl.Events = nil
	impl.TxEvents = nil
}

func (impl *LocalEventBusHandlerImpl) Run() error {
	registerAllHandler(impl.mockEventbus, impl.handlerConfigs)
	return impl.mockEventbus.Run()
}

func initEventbus() (tieventbus.Eventbus, error) {
	sqlDB, err := db.GetDB().DB()
	if err != nil {
		err = billing_errors.NewSystemError(err)
		return nil, err
	}
	res, err := tieventbus.NewEventbusMock(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 NewLocalEventBusHandler() (*LocalEventBusHandlerImpl, error) {
	eb, err := initEventbus()
	if err != nil {
		return nil, err
	}
	return &LocalEventBusHandlerImpl{
		mockEventbus: eb,
	}, nil
}

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

func (impl *LocalEventBusHandlerImpl) Close() error {
	impl.mockEventbus.Close()
	return nil
}