package consumer

import (
	"context"
	"joybuywork/internal/app"
	"errors"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/rs/xid"
	"gits.branchcn.com/backend/microcode/plugins/kafka"
	"time"
)

const (
	defaultConsumerNum = 5 //默认消费者数
)

type Receive func(session sarama.ConsumerGroupSession, message *sarama.ConsumerMessage)

type CParams struct {
	Topic     string
	Num       int
	Name      string
	GroupName string
	Config    *sarama.Config
}

type IConsumer interface {
	Receive(session sarama.ConsumerGroupSession, message *sarama.ConsumerMessage)
	Params() CParams
}

type Consumer struct {
	c  IConsumer
	cp CParams
}

func NewConsumer(consumer IConsumer) (*Consumer, error) {
	c := &Consumer{
		c:  consumer,
		cp: consumer.Params(),
	}

	if c.cp.Topic == "" {
		return nil, errors.New("topic empty")
	}

	if c.cp.GroupName == "" {
		return nil, errors.New("group_name empty")
	}

	if c.cp.Num == 0 {
		c.cp.Num = defaultConsumerNum
	}

	if c.cp.Config == nil {
		c.cp.Config = kafka.DefaultConsumerConfig()
	}

	if c.cp.Name == "" {
		c.cp.Name = xid.New().String()
	}

	return c, nil
}

func (c *Consumer) Run() {
	handler := newGroupHandler(c.c.Receive)
	for i := 0; i < c.cp.Num; i++ {
		idx := i
		app.GoSafeRetry(func() {
			start(handler, c.cp, idx)
		}, func() {
			app.Logger.Warn(context.Background(), "消费者将在5秒后尝试重新启动", "")
			time.Sleep(5 * time.Second)
			start(handler, c.cp, idx)
		}, "消费者组", c.cp.GroupName)
	}
	return
}

// 启动一个消费任务
func start(handler *GroupHandler, conf CParams, idx int) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 创建消费者组
	consumerGroup, err := app.Kafka.NewConsumerGroup(conf.GroupName, conf.Config)
	if err != nil {
		app.Logger.Error(ctx, fmt.Sprintf("创建消费者组出错:%v", err))
		return
	}
	defer consumerGroup.Close()

	//加入全局运行任务
	mark := fmt.Sprintf("%s-%d", conf.Name, idx)
	RTS[mark] = &Task{
		gh:     handler,
		cancel: cancel,
	}

	for {
		if err = consumerGroup.Consume(ctx, []string{conf.Topic}, handler); err != nil {
			app.Logger.Error(ctx, fmt.Sprintf("消费出错:%v", err))
		}
		if handler.Closed() {
			// 消费者处理程序被关闭，退出消费循环
			delete(RTS, mark)
			break
		}
		app.Logger.Warn(ctx, fmt.Sprintf("消费者(%s-%d)重新分配", conf.Name, idx))
	}
	app.Logger.Warn(ctx, fmt.Sprintf("消费者(%s-%d)结束消费", conf.Name, idx))
}