package eventbus

import (
	"context"
	"database/sql"
	"dataservices/commons/utils"
	"fmt"
	"time"

	"github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

const (
	DataserviceGroup = "dataservice"
)

type Event tieventbus.Event

type HandlerConfig struct {
	EventType string // the eventTypeKey you want to subscribe to
	// Group           string     // the consumer groupKey, one groupKey only process event once, and diff groups can consume one event repeatedly
	NumOfGoroutines uint // default 1
	RetryTimes      uint // default 0, disable retry
	RetryInterval   time.Duration
	HandleFunc      HandleFunc // Function called when event is received

	// is the instance of data structure of data.
	// Eventbus uses it by uses it by new a pointer of your struct,
	// usage: handleConfig.DataStruct = YourStruct{} or handleConfig.DataStruct = &YourStruct{}
	// support proto.Message
	DataStruct interface{}
}

type HandleFunc func(ctx context.Context, logger *zap.Logger, event *Event) error

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

	PublishEvent(ctx context.Context, event *Event) error

	PublishEventWithTX(ctx context.Context, event *Event, gTx *gorm.DB) error

	// event data should only be a struct not pointer, and can be cloned by golang := operator
	RegisterEventHandler(configs ...HandlerConfig) Eventbus
}

var global Eventbus

func Run() error {
	return Get().Run()
}

func Init(appName string, db *sql.DB, handlerConf ...HandlerConfig) Eventbus {
	bus, err := tieventbus.NewEventbus(tieventbus.EventbusConfig{
		AppName:    appName,
		DB:         db,
		TxPubTable: appName, // appName as eventbus table
		MQTimeout:  10 * time.Second,
	})
	if err != nil {
		panic(err)
	}
	global = &eventbusImpl{bus: bus}
	global.RegisterEventHandler(handlerConf...)
	return global
}

func Get() Eventbus {
	if global == nil {
		panic("Eventbus has not been initialized!!!")
	}
	return global
}

// for test only
func SetDefaultBus(bus Eventbus) {
	global = bus
}

type eventbusImpl struct {
	bus tieventbus.Eventbus
}

func (impl *eventbusImpl) getImpl() tieventbus.Eventbus {
	return impl.bus
}

func (impl *eventbusImpl) Run() error {
	tilog.Info(context.Background(), "Eventbus begin run...")
	return impl.getImpl().Run()
}

func (impl *eventbusImpl) PublishEvent(ctx context.Context, event *Event) error {
	tieventbusImpl := impl.getImpl()
	err := tieventbusImpl.Publish(&tieventbus.Event{
		ID:   event.ID,
		Type: event.Type,
		Data: event.Data,
	})
	if err != nil {
		tilog.Error(ctx, "Failed to publish event", zap.String("event_id", event.ID), zap.String("event_type", event.Type), zap.Error(err))
		err = errs.NewInternalMQ(fmt.Sprintf("Failed to publish event, event_id:%s, event_type:%s", event.ID, event.Type), err)
		return err
	}
	tilog.Info(ctx, "Succeed to publish event", zap.String("event_id", event.ID), zap.String("event_type", event.Type))
	return nil
}

func (impl *eventbusImpl) PublishEventWithTX(ctx context.Context, event *Event, gTx *gorm.DB) error {
	tieventbusImpl := impl.getImpl()
	tx := gTx.Statement.ConnPool.(*sql.Tx)

	err := tieventbusImpl.PublishWithDBTx(tx, &tieventbus.Event{
		ID:   event.ID,
		Type: event.Type,
		Data: event.Data,
	})
	if err != nil {
		tilog.Error(ctx, "Failed to publish event with tx", zap.Any("event", event), zap.Error(err))
		err = errs.NewInternalMQ(fmt.Sprintf("Failed to publish event, event_id:%s, event_type:%s", event.ID, event.Type), err)
		return err
	}
	tilog.Info(ctx, "Succeed to publish event with tx", zap.Any("event", event))
	return nil
}

func (impl *eventbusImpl) RegisterEventHandler(configs ...HandlerConfig) Eventbus {
	ctx := context.Background()
	for i := range configs {
		config := configs[i]
		tilog.Info(ctx, "Register event handler",
			zap.String("event_type", config.EventType),
			zap.Uint("num_of_goroutines", config.NumOfGoroutines),
			zap.Uint("retry_times", config.RetryTimes),
			zap.Duration("retry_interval", config.RetryInterval),
		)
		impl.getImpl().RegisterHandler(tieventbus.HandlerConfig{
			EventType: config.EventType,
			Group:     DataserviceGroup,
			HandleFunc: func(event *tieventbus.Event) error {
				ctx, logger, handleID := preHandleEvent((*Event)(event))
				return utils.NOPanicWrap("eventbus_handler_"+handleID, func() error {
					start := time.Now()
					err := config.HandleFunc(ctx, logger, (*Event)(event))
					logger.Info("eventbus handler completed", zap.String("cost", time.Since(start).String()), zap.Bool("succeed", err == nil), zap.Error(err))
					// if err != nil {
					// 	logger.Info("eventbus handler failed, wait a moment and retry", zap.Int("retry_times", int(config.RetryTimes)), zap.Duration("retry_interval", config.RetryInterval))
					// 	<-time.After(config.RetryInterval)
					// }
					return err
				})()
			},
			NumOfGoroutines: config.NumOfGoroutines,
			RetryTimes:      config.RetryTimes,
			DataStruct:      config.DataStruct,
		})
	}
	return impl
}

func preHandleEvent(event *Event) (context.Context, *zap.Logger, string) {
	ctx := context.Background()
	handleID := event.ID
	if handleID == "" {
		handleID = utils.RandStr(32)
	}
	logger := tilog.WithContextL(ctx).With(zap.String("event_id", handleID)).With(zap.String("event_type", event.Type))
	logger.Info("eventbus handler got new event", zap.Any("event", event))
	return ctx, logger, handleID
}