package mq

import (
	"context"
	"encoding/json"
	"gitee.com/h79/goutils/common/logger"
	"gitee.com/h79/goutils/common/result"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	prod "github.com/apache/rocketmq-client-go/v2/producer"
	"go.uber.org/zap"
)

var _ Producer = (*producerEx)(nil)

type producerEx struct {
	cfg      Config
	producer rocketmq.Producer
}

func NewProducer(cfg Config) Producer {
	return &producerEx{cfg: cfg}
}

func (rp *producerEx) Start() error {

	p, e := rocketmq.NewProducer(
		prod.WithNsResolver(primitive.NewPassthroughResolver([]string{rp.cfg.Client.Server})),
		prod.WithRetry(rp.cfg.ProducerConf.Retry),
		prod.WithQueueSelector(prod.NewManualQueueSelector()))
	if e != nil {
		return result.Errorf(result.ErrMqInitInternal, "[MQ] Init producer failed, error: %v", e).Log()
	}

	if e := p.Start(); e != nil {
		return result.Errorf(result.ErrMqStartInternal, "[MQ] Start producer failed, error: %v", e).Log()
	}
	rp.producer = p
	return nil
}

func (rp *producerEx) Stop() {
	_ = rp.producer.Shutdown()
}

func (rp *producerEx) SendJson(topic string, event interface{}) error {
	body, _ := json.Marshal(event)
	return rp.SendBytes(topic, body)
}

func (rp *producerEx) SendBytes(topic string, data []byte) error {

	logger.L().Debug("MQ",
		zap.String("topic", topic),
		zap.ByteString("data", data))
	ctx := context.Background()
	if _, err := rp.producer.SendSync(ctx, primitive.NewMessage(topic, data)); err != nil {
		return result.Errorf(result.ErrMqPublishInternal, "MQ: Publish String topic(%s) failure, err= %+v", topic, err).Log()
	}
	return nil
}

func (rp *producerEx) SendAsync(topic string, data []byte) error {
	logger.L().Debug("MQ",
		zap.String("topic", topic),
		zap.ByteString("data", data))
	ctx := context.Background()
	if err := rp.producer.SendAsync(ctx, func(ctx context.Context, result *primitive.SendResult, err error) {

	}, primitive.NewMessage(topic, data)); err != nil {
		return result.Errorf(result.ErrMqPublishInternal, "MQ: Publish String topic(%s) failure, err= %+v", topic, err).Log()
	}
	return nil
}