package tieventbus

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"strconv"
	"sync"
	"time"

	"github.com/ThreeDotsLabs/watermill/components/forwarder"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/ThreeDotsLabs/watermill/message/router/middleware"
	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/tiinf/transport/tieventbus/metadata"
	"github.com/tidbcloud/tiinf/transport/tieventbus/metrics"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
)

const (
	SourcePublisher = "publisher"
	SourceConsumer  = "consumer"
	SourceDB        = "db"
	SourceMQ        = "mq"

	TargetMQ  = "mq"
	TargetDB  = "db"
	TargetDLQ = "dlq"

	DupGroupEventTypeConsuming = "can not consume 1 eventType with same group, please check you registering"
)

func messageFromEvent(event *Event) (*message.Message, error) {
	var payload []byte
	var err error
	marshallType := metadata.MarshalTypeJSON
	protoMsg, ok := event.Data.(proto.Message)
	if ok {
		marshallType = metadata.MarshalTypeProto
	}
	switch marshallType {
	case metadata.MarshalTypeJSON:
		payload, err = json.Marshal(event.Data)
		break
	case metadata.MarshalTypeProto:
		payload, err = protojson.Marshal(protoMsg)
		break
	default:
		err = errors.New("unknown MarshalType")
		break
	}
	if err != nil {
		return nil, err
	}
	msg := message.NewMessage(event.ID, payload)
	msg.Metadata.Set(metadata.EventTypeKey, event.Type)
	msg.Metadata.Set(metadata.MarshalTypeKey, strconv.FormatUint(uint64(marshallType), 10))
	ctx := CtxWithEventTypeValue(msg.Context(), event.Type)
	msg.SetContext(ctx)
	return msg, nil
}

func dataTypeOf(dataStruct interface{}) (reflect.Type, error) {
	tp := reflect.TypeOf(dataStruct)
	if tp.Kind() == reflect.Ptr {
		tp = tp.Elem()
	}
	if tp.Kind() == reflect.Ptr {
		return nil, errors.New("eventbus data do not support pointer to pointer")
	}
	return tp, nil
}

func eventFromMessage(eventType string, tp reflect.Type) func(*message.Message) (*Event, error) {
	_, ok := reflect.New(tp).Interface().(proto.Message)
	marshallType := metadata.MarshalTypeJSON
	if ok {
		marshallType = metadata.MarshalTypeProto
	}
	return func(msg *message.Message) (event *Event, err error) {
		ptr := reflect.New(tp).Interface()
		switch marshallType {
		case metadata.MarshalTypeJSON:
			deco := json.NewDecoder(bytes.NewBuffer(msg.Payload))
			deco.UseNumber()
			err = deco.Decode(ptr)
			break
		case metadata.MarshalTypeProto:
			err = protojson.Unmarshal(msg.Payload, ptr.(proto.Message))
			break
		default:
			err = errors.New("unknown marshal type")
			break
		}
		if err != nil {
			return nil, err
		}
		return &Event{
			ID:   msg.UUID,
			Type: eventType,
			Data: ptr,
		}, nil
	}
}

func getDLQTopic(topic string) string {
	return fmt.Sprintf("%s-dlq", topic)
}

func (ec *EventbusConfig) validate() error {
	if ec.AppName == "" {
		return fmt.Errorf("AppName should not be empty")
	}
	return nil
}

func (ec *EventbusConfig) setDefault() {
	if ec.DB != nil && ec.TxPubTable == "" {
		ec.TxPubTable = "eventbus_tx_pub_table"
	}
	if ec.MQTimeout == 0 {
		ec.MQTimeout = time.Second
	}
}

func (hc *HandlerConfig) validate() error {
	if hc.EventType == "" {
		return fmt.Errorf("EventType should not be empty")
	}
	if hc.Group == "" {
		return fmt.Errorf("Group should not be empty")
	}
	return nil
}

func (hc *HandlerConfig) setDefault() {
	if hc.NumOfGoroutines == 0 {
		hc.NumOfGoroutines = 1
	}
}

type EventbusImpl struct {
	config         *EventbusConfig
	registered     []HandlerConfig
	pubSub         pubSub
	running        bool
	runningChan    chan struct{}
	closed         bool
	closedChan     chan struct{}
	lock           *sync.Mutex
	router         *message.Router
	fwd            *forwarder.Forwarder
	metricsBuilder *metrics.PrometheusMetricsBuilder
}

func handlerFuncWrapper(config HandlerConfig) (message.HandlerFunc, error) {
	tp, err := dataTypeOf(config.DataStruct)
	if err != nil {
		return nil, err
	}
	return func(msg *message.Message) (messages []*message.Message, err error) {
		ctx := msg.Context()
		retryTimes := uint64(0)
		if ctx.Value(retryTimesKey) != nil {
			retryTimes = RetryTimesFromCtx(ctx)
			retryTimes++
		}
		defer func() {
			ctx := msg.Context()
			ctx = CtxWithRetryTimes(ctx, retryTimes)
			ctx = CtxWithEventTypeValue(ctx, config.EventType)
			ctx = CtxWithGroupValue(ctx, config.Group)
			ctx = CtxWithSourceValue(ctx, SourceMQ)
			ctx = CtxWithSuccessValue(ctx, err == nil)
			msg.SetContext(ctx)
		}()
		event, err := eventFromMessage(config.EventType, tp)(msg)
		if err != nil {
			return nil, err
		}
		return nil, config.HandleFunc(event)
	}, nil
}

func (e *EventbusImpl) enableHandlers(fixDLQ bool) error {
	publisher, err := e.pubSub.GetMQPublisher()
	if err != nil {
		return err
	}
	publisher, err = e.publisherWrapMetrics(publisher)
	if err != nil {
		return err
	}
	dedup := make(map[string]bool)
	for _, h := range e.registered {
		h.setDefault()
		err := h.validate()
		if err != nil {
			return err
		}
		group := fmt.Sprintf("%s-%s", e.config.AppName, h.Group)
		sub, err := e.pubSub.GetMQSubscriber(group)
		if err != nil {
			return err
		}
		poisonQueueTopic := getDLQTopic(h.EventType)
		topic := h.EventType
		if fixDLQ {
			topic = poisonQueueTopic
		}
		handlerName := fmt.Sprintf("%s-%s-handler", h.EventType, group)
		if _, ok := dedup[handlerName]; ok {
			return fmt.Errorf(DupGroupEventTypeConsuming)
		}
		dedup[handlerName] = true
		handlerFunc, err := handlerFuncWrapper(h)
		if err != nil {
			return err
		}
		for i := uint(0); i < h.NumOfGoroutines; i++ {
			handlerName := fmt.Sprintf("%s-%d", handlerName, i)
			handler := e.router.AddHandler(
				handlerName,
				topic,
				sub,
				"",
				disabledPublisher{},
				handlerFunc,
			)
			var middlewares []message.HandlerMiddleware
			if !fixDLQ {
				dlqMiddleware, err := deadLetterQueueMiddleware(publisher, h)
				if err != nil {
					return err
				}
				middlewares = append(middlewares, dlqMiddleware)
			}
			middlewares = append(middlewares, retryMiddleware(h.RetryTimes), middleware.Recoverer)
			handler.AddMiddleware(middlewares...)
		}
	}
	return nil
}

func (e *EventbusImpl) RegisterHandler(config HandlerConfig) {
	e.registered = append(e.registered, config)
}

func (e *EventbusImpl) Publish(event *Event) error {
	if !e.running {
		return fmt.Errorf("eventbus is not running")
	}
	msg, err := messageFromEvent(event)
	if err != nil {
		return err
	}
	publisher, err := e.pubSub.GetMQPublisher()
	if err != nil {
		return err
	}
	publisher, err = publisherCtxDecorator(SourcePublisher, TargetMQ)(publisher)
	if err != nil {
		return err
	}
	publisher, err = e.publisherWrapMetrics(publisher)
	if err != nil {
		return err
	}
	return publisher.Publish(event.Type, msg)
}

func (e *EventbusImpl) PublishAsync(event *Event) *AsyncResult {
	res := &AsyncResult{
		Finished: make(chan struct{}),
	}
	go func() {
		var err error
		defer func() {
			res.Error = err
			close(res.Finished)
		}()
		err = e.Publish(event)
	}()
	return res
}

func (e *EventbusImpl) PublishWithDBTx(tx *sql.Tx, event *Event) error {
	if !e.running {
		return fmt.Errorf("eventbus is not running")
	}
	if e.config.DB == nil {
		return fmt.Errorf("failed to publish with transaction on EventbusConfig.DB is nil")
	}
	msg, err := messageFromEvent(event)
	if err != nil {
		return err
	}
	forwarderPublisher, err := e.newForwarderPublisher(tx)
	if err != nil {
		return err
	}
	publisher, err := publisherCtxDecorator(SourcePublisher, TargetDB)(forwarderPublisher)
	if err != nil {
		return err
	}
	publisher, err = e.publisherWrapMetrics(publisher)
	if err != nil {
		return err
	}
	return publisher.Publish(event.Type, msg)
}

func (e *EventbusImpl) Running() chan struct{} {
	if len(e.registered) != 0 {
		return e.router.Running()
	}
	return e.runningChan
}

func (e *EventbusImpl) Close() {
	if !e.running {
		return
	}
	e.closed = true
	e.running = false
	if e.fwd != nil {
		err := e.fwd.Close()
		if err != nil {
			log.Error("Failed to close Forwarder", zap.Error(err))
		}
	}
	if len(e.registered) != 0 && e.router != nil {
		err := e.router.Close()
		if err != nil {
			log.Error("Failed to close Router", zap.Error(err))
		}
	}
	close(e.closedChan)
}

func (e *EventbusImpl) run() (err error) {
	if e.closed {
		return fmt.Errorf("eventbus is closed")
	}
	if e.running {
		return nil
	}
	e.running = true
	if e.metricsBuilder != nil {
		e.router.AddMiddleware(e.metricsBuilder.NewRouterMiddleware().Middleware)
		e.router.AddSubscriberDecorators(e.metricsBuilder.DecorateSubscriber)
	}
	err = e.enableHandlers(false)
	if err != nil {
		return err
	}
	err = e.enableForwarder()
	if err != nil {
		return err
	}
	if len(e.registered) == 0 {
		return e.wait()
	}
	return e.router.Run(context.Background())
}

func (e *EventbusImpl) Run() (err error) {
	e.lock.Lock()
	defer e.lock.Unlock()
	failed := make(chan struct{})
	go func() {
		err = e.run()
		if err != nil {
			close(failed)
		}
	}()
	select {
	case <-e.Running():
		return nil
	case <-failed:
		return err
	case <-time.After(10 * time.Second):
		return fmt.Errorf("run eventbus timeout")
	}
}

func (e *EventbusImpl) RunUtilClosed() (err error) {
	e.lock.Lock()
	defer e.lock.Unlock()
	return e.run()
}

func (e *EventbusImpl) FixDLQ(timeout time.Duration) error {
	e.lock.Lock()
	defer e.lock.Unlock()
	if e.closed {
		return fmt.Errorf("eventbus is closed")
	}
	if len(e.registered) == 0 {
		return fmt.Errorf("can not fix dlq on registered handlers is empty")
	}
	defer e.Close()
	e.running = true
	var err error
	handlerCalled := make(chan bool)
	e.router.AddMiddleware(errorAsPanicMiddleware, handlerCalledMiddleware(handlerCalled))
	err = e.enableHandlers(true)
	if err != nil {
		return err
	}
	err = e.enableForwarder()
	if err != nil {
		return err
	}
	go func() {
		log.Info("Start fixDLQ router")
		err := e.router.Run(context.Background())
		if err != nil {
			log.Error("Failed to run Router", zap.Error(err))
		}
	}()
	select {
	case <-e.Running():
		return e.waitFixDLQFinished(handlerCalled, timeout)
	case <-time.After(time.Second * 60):
		return fmt.Errorf("timeout for wait router start")
	}
}

func (e *EventbusImpl) wait() error {
	close(e.runningChan)
	<-e.closedChan
	return nil
}

func (e *EventbusImpl) waitFixDLQFinished(handlerCalled chan bool, timeout time.Duration) error {
	for {
		select {
		case success := <-handlerCalled:
			if !success {
				log.Error("Error occurred on fixDQL, try to close eventbus")
				return nil
			}
		case <-time.After(timeout):
			log.Info("Timed out for no message consuming, try to close eventbus.", zap.Duration("timeout", timeout))
			return nil
		}
	}
}

func newEventbus(conf EventbusConfig, mock bool) (Eventbus, error) {
	err := conf.validate()
	if err != nil {
		return nil, err
	}
	conf.setDefault()
	var ps pubSub
	if mock {
		ps, err = newPubSubMock()
	} else {
		ps, err = newPubSubImpl(conf)
	}
	if err != nil {
		return nil, err
	}
	router, err := newRouter()
	if err != nil {
		return nil, err
	}
	eb := &EventbusImpl{
		config:      &conf,
		pubSub:      ps,
		router:      router,
		lock:        &sync.Mutex{},
		closed:      false,
		closedChan:  make(chan struct{}),
		running:     false,
		runningChan: make(chan struct{}),
	}
	return eb, nil
}

func NewEventbus(conf EventbusConfig) (Eventbus, error) {
	return newEventbus(conf, false)
}

func NewEventbusMock(conf EventbusConfig) (Eventbus, error) {
	return newEventbus(conf, true)
}

func (e *EventbusImpl) EnableMetrics(prometheusRegistry prometheus.Registerer, namespace string, subsystem string) {
	metricsBuilder := metrics.NewPrometheusMetricsBuilder(prometheusRegistry, namespace, subsystem)
	e.metricsBuilder = &metricsBuilder
	labelGetters := map[string]func(context.Context) string{
		metrics.LabelKeyEventType: EventTypeFromCtx,
		metrics.LabelKeyGroup:     GroupFromCtx,
		metrics.LabelKeySource:    SourceFromCtx,
		metrics.LabelKeyTarget:    TargetFromCtx,
		metrics.LabelKeyRetryTimes: func(ctx context.Context) string {
			return strconv.FormatUint(RetryTimesFromCtx(ctx), 10)
		},
		metrics.LabelKeySuccess: func(ctx context.Context) string {
			return strconv.FormatBool(SuccessFromCtx(ctx))
		},
	}
	metrics.SetLabelGetters(labelGetters)
}