package tieventbus

import (
	"context"
	"database/sql"
	"fmt"
	"strconv"
	"time"

	"github.com/ThreeDotsLabs/watermill/components/forwarder"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/pingcap/log"
	"github.com/tidbcloud/tiinf/transport/tieventbus/metadata"
	"go.uber.org/zap"
)

func (e *EventbusImpl) newForwarder() (*forwarder.Forwarder, error) {
	db := e.config.DB
	forwarderTopic := e.config.TxPubTable
	kafkaPublisher, err := e.pubSub.GetMQPublisher()
	if err != nil {
		return nil, err
	}
	publisher, err := publisherCtxDecorator(SourceDB, TargetMQ)(kafkaPublisher)
	if err != nil {
		return nil, err
	}
	publisher, err = e.publisherWrapMetrics(publisher)
	if err != nil {
		return nil, err
	}
	publisher, err = message.MessageTransformPublisherDecorator(func(m *message.Message) {
		m.SetContext(CtxWithEventTypeValue(m.Context(), m.Metadata.Get(eventTypeKey)))
	})(publisher)
	if err != nil {
		return nil, err
	}
	sqlSubscriber, err := getSQLSubscriber(db)
	if err != nil {
		return nil, err
	}
	subscriber, err := message.MessageTransformSubscriberDecorator(func(m *message.Message) {
		ctx := CtxWithEventTypeValue(m.Context(), e.config.TxPubTable)
		ctx = CtxWithGroupValue(ctx, "nil")
		ctx = CtxWithSourceValue(ctx, SourceDB)
		m.SetContext(ctx)
	})(sqlSubscriber)
	if err != nil {
		return nil, err
	}
	subscriber, err = e.subscriberWrapMetrics(subscriber)
	if err != nil {
		return nil, err
	}
	var middlewares []message.HandlerMiddleware
	if e.metricsBuilder != nil {
		middlewares = append(middlewares, e.metricsBuilder.NewRouterMiddleware().Middleware)
	}
	middlewares = append(
		middlewares,
		func(h message.HandlerFunc) message.HandlerFunc {
			return func(msg *message.Message) (messages []*message.Message, err error) {
				defer func() {
					msg.SetContext(CtxWithSuccessValue(msg.Context(), err == nil))
				}()
				return h(msg)
			}
		},
	)
	return forwarder.NewForwarder(subscriber, publisher, logger, forwarder.Config{
		ForwarderTopic: forwarderTopic,
		Middlewares:    middlewares,
	})
}

func (e *EventbusImpl) newForwarderPublisher(tx *sql.Tx) (message.Publisher, error) {
	forwarderTopic := e.config.TxPubTable
	sqlPublisher, err := getSQLTxPublisher(tx)
	if err != nil {
		return nil, err
	}
	publisher, err := message.MessageTransformPublisherDecorator(func(m *message.Message) {
		m.Metadata.Set(metadata.CreatedAtKey, strconv.FormatInt(time.Now().UTC().UnixNano(), 10))
	})(sqlPublisher)
	if err != nil {
		return nil, err
	}
	forwarderPublisher := forwarder.NewPublisher(publisher, forwarder.PublisherConfig{
		ForwarderTopic: forwarderTopic,
	})
	return forwarderPublisher, nil
}

func (e *EventbusImpl) enableForwarder() error {
	if e.config.DB == nil {
		return nil
	}
	fwd, err := e.newForwarder()
	if err != nil {
		return err
	}
	e.fwd = fwd
	go func() {
		err := fwd.Run(context.Background())
		if err != nil {
			log.Error("forwarder exit on error", zap.Error(err))
		}
	}()
	select {
	case <-fwd.Running():
		return nil
	case <-time.After(10 * time.Second):
		return fmt.Errorf("forwarder start timeout")
	}
}