joybuywork/internal/service/queue/consumer/campaignnotice/consume.go

package campaignnotice

import (
	"joybuywork/internal/app"
	"joybuywork/internal/consts/ckey"
	"joybuywork/internal/service/queue/consumer"
	"joybuywork/internal/service/queue/consumer/campaignnotice/export"
	"joybuywork/internal/service/queue/consumer/campaignnotice/validevents"
	"joybuywork/internal/service/queue/producer"
	"joybuywork/internal/service/queue/types"
	"joybuywork/internal/utils"
	"joybuywork/internal/utils/logic"
	"errors"

	"github.com/Shopify/sarama"
	"gits.branchcn.com/backend/microcode/plugins/derivative/task"
	"gits.branchcn.com/backend/microcode/plugins/kafka"
)

func Run() (err error) {
	c, err := consumer.NewConsumer(&CampaignNotice{})
	if err != nil {
		return
	}
	c.Run()
	return
}

type CampaignNotice struct{}

func (s CampaignNotice) Params() consumer.CParams {
	return consumer.CParams{
		Topic:     types.CampaignNoticeTopic,
		Num:       5,
		Name:      "campaign_notice",
		GroupName: types.CampaignUserGroup,
		Config:    kafka.DefaultConsumerConfig(),
	}
}

func (s CampaignNotice) Receive(session sarama.ConsumerGroupSession, message *sarama.ConsumerMessage) {
	// 结束标记消息已消费
	defer session.MarkMessage(message, "")
	var handler *consumer.Ctx
	switch utils.Bs(message.Key) {
	case types.CampaignNoticeKey:
		handler = consumer.NewCtx("活动通知", message, campaignNotice)
	case types.CampaignOtherStatKey:
		handler = consumer.NewCtx("活动其他统计", message, campaignOtherStat)
	case types.CampaignRewardNotifyKey:
		handler = consumer.NewCtx("奖品回调", message, campaignRewardNotify)
	case types.CampaignTargetStatKey:
		handler = consumer.NewCtx("用户达标统计", message, campaignTargetStatKey)
	case types.CampaignCondStatKey:
		handler = consumer.NewCtx("活动条件统计", message, campaignCondStatKey)
	case types.CampaignBehaviorKey:
		handler = consumer.NewCtx("活动行为处理", message, campaignBehavior)
	case types.CampaignDiceMapKey:
		handler = consumer.NewCtx("大富翁地图生成", message, campaignDiceMap)
	case types.DownloadKey:
		handler = consumer.NewCtx("导出任务", message, download)
	case types.ValidEventsKey:
		handler = consumer.NewCtx("有效事件", message, validEvents)
	case types.ProjectUserKey:
		handler = consumer.NewCtx("项目用户存储", message, projectUser)
	case types.RewardLimitUpKey:
		handler = consumer.NewCtx("活动修改预算总上线", message, rewardLimitUp)
	case types.CampaignChildInviteKey:
		handler = consumer.NewCtx("子活动排行榜结算", message, campaignChildInvite)
	case types.ProjectUserCoinDeductKey:
		handler = consumer.NewCtx("项目用户积分扣除", message, projectUserCoinDeduct)
	case types.CampaignVsSettlementKey:
		handler = consumer.NewCtx("赛事竞猜结算", message, campaignVsSettlement)
	case types.CampaignSettlementRewardKey:
		handler = consumer.NewCtx("赛事竞猜结算奖励", message, campaignSettlementReward)
	case types.GameActionKey:
		handler = consumer.NewCtx("游戏行为类型", message, gameAction)
	default:
		return
	}

	handler.Done()
}

func gameAction(ctx *consumer.Ctx) (err error) {
	var val types.GameAction
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(GameAction)
	{
		container.Init(ctx.Context())
		container.Req = &val
	}

	return logic.Run(container)
}

func campaignSettlementReward(ctx *consumer.Ctx) (err error) {
	var val types.CampaignSettlementReward
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(SettlementReward)
	{
		container.Init(ctx.Context())
		container.Req = &val
	}

	return logic.Run(container)
}

func campaignVsSettlement(ctx *consumer.Ctx) (err error) {
	var val types.CampaignVsSettlement
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(VsSettlement)
	{
		container.Init(ctx.Context())
		container.Req = &val
	}

	return logic.Run(container)
}

func projectUserCoinDeduct(ctx *consumer.Ctx) (err error) {
	var val types.ProjectUserCoinDeduct
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	taskIdKey := ckey.ProjectUserCoinDeduct(val.UserId)
	t := app.NewRedisTask(ctx.Context(), taskIdKey)
	c, err := t.TaskPublicAndConsume(func(td *task.Data) error {
		var data types.ProjectUserCoinDeduct
		err = td.ParsingMsg(&data)
		if err != nil {
			return err
		}
		var container = new(ProjectUserCoinDeduct)
		{
			container.Init(td.Ctx)
			container.Req = &data
		}

		return container.Run()
	}, val)

	if err != nil {
		return
	}

	if c.Error() != nil && (errors.Is(c.Error(), task.AbnormalErr) || errors.Is(c.Error(), task.ReenterLimit)) {
		producer.NewCampaignNotice(ctx.Context()).ProjectUserCoinDeduct(val)
		err = c.Error()
		return
	}

	return
}

func campaignChildInvite(ctx *consumer.Ctx) (err error) {
	var val types.CampaignChildInvite
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(ChildInvite)
	{
		container.Init(ctx.Context())
		container.Req = &val
	}

	return logic.Run(container)
}

func projectUser(ctx *consumer.Ctx) (err error) {
	var val types.ProjectUser
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(ProjectUser)
	{
		container.Init(ctx.Context())
		container.Req = &val
	}

	return logic.Run(container)
}

func validEvents(ctx *consumer.Ctx) (err error) {
	var val types.ValidEvents
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	taskIdKey := ckey.ValidEventsKey(val.UserId)
	if val.NoticeType == types.NoticeTypeClick {
		taskIdKey = ckey.ValidEventsUserMarkKey(val.ClickInfo.UserMark)
	}

	t := app.NewRedisTask(ctx.Context(), taskIdKey)
	c, err := t.TaskPublicAndConsume(func(td *task.Data) error {
		var data types.ValidEvents
		err = td.ParsingMsg(&data)
		if err != nil {
			return err
		}
		var container = new(validevents.ValidEvents)
		{
			container.Init(td.Ctx, logic.WithBp(), logic.WithErrData(data))
			container.Req = &data
		}

		return logic.Run(container)
	}, val)

	if err != nil {
		return
	}

	if c.Error() != nil && (errors.Is(c.Error(), task.AbnormalErr) || errors.Is(c.Error(), task.ReenterLimit)) {
		producer.NewCampaignNotice(ctx.Context()).ValidEvents(val)
		err = c.Error()
		return
	}

	return
}

func download(ctx *consumer.Ctx) (err error) {
	var val types.Download
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(export.Download)
	{
		container.Init(ctx.Context(), logic.WithErrData(val))
		container.Req = &val
	}

	return logic.Run(container)
}

func campaignCondStatKey(ctx *consumer.Ctx) (err error) {
	var val types.CampaignCondStat
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(CondStat)
	{
		container.Init(ctx.Context())
		container.Req = &val
	}

	return logic.Run(container)
}

func campaignTargetStatKey(ctx *consumer.Ctx) (err error) {
	var val types.CampaignTargetStat
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(TargetStat)
	{
		container.Init(ctx.Context())
		container.Req = &val
	}

	return logic.Run(container)
}

func campaignRewardNotify(ctx *consumer.Ctx) (err error) {
	var val types.CampaignRewardNotify
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(RewardNotify)
	{
		container.Init(ctx.Context())
		container.Req = &val
	}

	return logic.Run(container)
}

func campaignNotice(ctx *consumer.Ctx) (err error) {
	var val types.CampaignNotice
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(Notice)
	{
		container.Init(ctx.Context(), logic.WithBp())
		container.Req = &val
	}

	return logic.Run(container)
}

func campaignOtherStat(ctx *consumer.Ctx) (err error) {
	var val types.CampaignOtherStat
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(OtherStat)
	{
		container.Init(ctx.Context())
		container.Req = &val
	}

	return logic.Run(container)
}

func campaignBehavior(ctx *consumer.Ctx) (err error) {
	var val types.CampaignBehavior
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(Behavior)
	{
		container.Init(ctx.Context())
		container.Req = &val
	}

	return logic.Run(container)
}

func campaignDiceMap(ctx *consumer.Ctx) (err error) {
	var val types.CampaignDiceMap
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(DiceMap)
	{
		container.Init(ctx.Context())
		container.Req = &val
	}

	return logic.Run(container)
}

func rewardLimitUp(ctx *consumer.Ctx) (err error) {
	var val types.RewardLimitUp
	err = ctx.Parse(&val)
	if err != nil {
		return
	}

	var container = new(RewardLimitUp)
	{
		container.Init(ctx.Context())
		container.Req = &val
	}

	return logic.Run(container)
}