package tieventbus

import (
	"time"

	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/ThreeDotsLabs/watermill/message/router/middleware"
	"github.com/hashicorp/go-multierror"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"go.uber.org/zap"
)

const (
	ReasonForPoisonedKey = "reason_poisoned"
)

func errorAsPanicMiddleware(h message.HandlerFunc) message.HandlerFunc {
	return func(event *message.Message) ([]*message.Message, error) {
		events, err := h(event)
		if err != nil {
			panic(err)
		}
		return events, nil
	}
}

func handlerCalledMiddleware(called chan bool) message.HandlerMiddleware {
	return func(h message.HandlerFunc) message.HandlerFunc {
		return func(event *message.Message) (events []*message.Message, err error) {
			defer func() {
				called <- err == nil
			}()
			return h(event)
		}
	}
}

func retryMiddleware(retryTimes uint) message.HandlerMiddleware {
	if retryTimes == 0 {
		return func(h message.HandlerFunc) message.HandlerFunc {
			return h
		}
	}
	retry := middleware.Retry{
		MaxRetries:      int(retryTimes),
		InitialInterval: time.Millisecond * 100,
		MaxInterval:     time.Millisecond * 100,
		Multiplier:      1.0,
		Logger:          logger,
	}
	return retry.Middleware
}

type poisonQueue struct {
	topic string
	pub   message.Publisher
}

// PoisonQueue provides a middleware that salvages unprocessable messages and published them on a separate topic.
// The main middleware chain then continues on, business as usual.
func newPoisonQueue(pub message.Publisher, topic string) (message.HandlerMiddleware, error) {
	if topic == "" {
		return nil, middleware.ErrInvalidPoisonQueueTopic
	}

	pq := poisonQueue{
		topic: topic,
		pub:   pub,
	}

	return pq.Middleware, nil
}

func (pq poisonQueue) publishPoisonMessage(msg *message.Message, err error) error {
	// no problems encountered, carry on
	if err == nil {
		return nil
	}

	// add context why it was poisoned
	msg.Metadata.Set(ReasonForPoisonedKey, err.Error())
	msg.Metadata.Set(eventTypeKey, EventTypeFromCtx(msg.Context()))
	msg.Metadata.Set(groupKey, GroupFromCtx(msg.Context()))

	// don't intercept error from publish. Can't help you if the publisher is down as well.
	return pq.pub.Publish(pq.topic, msg)
}

func (pq poisonQueue) Middleware(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) (events []*message.Message, err error) {
		defer func() {
			if err != nil {
				// handler didn't cope with the message; publish it on the poison topic and carry on as usual
				log.Error("Error occurred. Send poison event to DLQ", zap.Error(err), zap.String("EventType", EventTypeFromCtx(msg.Context())), zap.String("Group", GroupFromCtx(msg.Context())))
				sendMsg := msg.Copy()
				sendMsg.SetContext(msg.Context())
				publishErr := pq.publishPoisonMessage(sendMsg, err)
				if publishErr != nil {
					publishErr = errors.Wrap(publishErr, "cannot publish message to poison queue")
					err = multierror.Append(err, publishErr)
					return
				}
				msg.SetContext(CtxWithSuccessValue(msg.Context(), false))
				err = nil
				return
			}
		}()
		// if h fails, the deferred function will salvage all that it can
		return h(msg)
	}
}

func deadLetterQueueMiddleware(publisher message.Publisher, config HandlerConfig) (message.HandlerMiddleware, error) {
	dlqTopic := getDLQTopic(config.EventType)
	publisher, err := publisherCtxDecorator(SourceConsumer, TargetDLQ)(publisher)
	if err != nil {
		return nil, err
	}
	poisonQueueMiddleware, _ := newPoisonQueue(publisher, dlqTopic)
	return poisonQueueMiddleware, err
}