package consumer

import (
	"context"
	"joybuywork/internal/app"
	"joybuywork/internal/consts"
	"joybuywork/internal/plugins/jsonm"
	"joybuywork/internal/plugins/printer"
	"joybuywork/internal/utils"
	"fmt"
	"time"

	"github.com/Shopify/sarama"
	"github.com/rs/xid"
)

type Do func(ctx *Ctx) error

type Ctx struct {
	ctx     context.Context
	name    string
	message *sarama.ConsumerMessage
	do      Do
}

func NewCtx(name string, message *sarama.ConsumerMessage, do Do) *Ctx {
	ctx := context.WithValue(context.Background(), consts.CtxMsgId, xid.New().String())
	for _, header := range message.Headers {
		ctx = context.WithValue(ctx, utils.Bs(header.Key), utils.Bs(header.Value))
	}

	return &Ctx{
		ctx:     ctx,
		name:    name,
		message: message,
		do:      do,
	}
}

func (s *Ctx) Done() {
	if s == nil {
		return
	}

	req := utils.Bs(s.message.Value)
	defer app.TryCatch(s.ctx, fmt.Sprintf("执行任务[%s]失败，数据: %s", s.name, req))

	var err error
	tn := time.Now()
	sl := printer.NewSlowLog(s.ctx, printer.SetSlow(printer.TenNormalSlow))
	app.Collect(s.name)
	sl.Start(fmt.Sprintf("运行[%s], 执行间隔：%12v, 数据：%s", s.name, time.Since(s.message.Timestamp), req))
	defer func() {
		sl.End(fmt.Sprintf("执行时长：%12v, 运行错误：%v", time.Since(tn), err), err, s.name, req, "")
	}()

	err = s.do(&Ctx{
		ctx:     s.ctx,
		message: s.message,
	})
}

func (s *Ctx) Parse(data any) error {
	return jsonm.Unmarshal(s.message.Value, data)
}

func (s *Ctx) Context() context.Context {
	return s.ctx
}