joybuywork/internal/service/crons/action/notify.go

package action

import (
	"joybuywork/internal/app"
	"joybuywork/internal/consts/ckey"
	"joybuywork/internal/repository/dao"
	"joybuywork/internal/repository/model"
	"joybuywork/internal/service/crons/action/containers"
	"joybuywork/internal/service/queue/producer"
	"joybuywork/internal/service/queue/types"
	"joybuywork/internal/utils"
	"joybuywork/internal/utils/logic"
	"joybuywork/proto/go_proto/workerr"
	"sync"
	"time"

	"github.com/panjf2000/ants/v2"
)

type Notify struct {
	Action
}

func (s *Notify) Desc() string {
	return "重新回调"
}

func (s *Notify) OnceKey() *OnceKey {
	return &OnceKey{
		exp:     2 * time.Minute,
		key:     ckey.CronNotifyReward(),
		release: true,
	}
}

func (s *Notify) Exec() (err error) {
	m := dao.FissionNotifyTask
	ntfM := dao.FissionNotify

	//找到需要重新回调的任务
	list, err := m.WithContext(s.ctx).Where(m.DoTime.Lte(time.Now().Add(-time.Hour).Unix())).Find()
	if err != nil {
		return
	}

	if len(list) == 0 {
		return
	}

	//任务对应的回调数据
	var ntfIds []int64
	for _, v := range list {
		ntfIds = append(ntfIds, v.NotifyID)
	}

	var idsGroup = utils.Groups[int64](ntfIds, 200)
	gop, _ := ants.NewPool(10)
	defer gop.Release()

	wg := sync.WaitGroup{}
	for _, ids := range idsGroup {
		wg.Add(1)
		_ = gop.Submit(func() {
			defer wg.Done()
			ntfL, _ := ntfM.WithContext(s.ctx).Where(ntfM.ID.In(ids...)).Find()

			var ntfMap = make(map[int64]*model.FissionNotify)
			for _, v := range ntfL {
				ntfMap[v.ID] = v
			}

			for _, v := range list {
				producer.RewardPushPub(s.ctx, 1, types.DelayCommonRewardPush{TaskId: v.ID})
			}
		})
	}

	wg.Wait()

	return
}

type PushNotify struct {
	Action
}

func (s *PushNotify) Desc() string {
	return "推送提醒回调"
}

func (s *PushNotify) OnceKey() *OnceKey {
	return &OnceKey{
		exp:     2 * time.Minute,
		key:     ckey.CronPushNotify(),
		release: true,
	}
}

func (s *PushNotify) Exec() (err error) {
	container := new(containers.PushNotify)
	{
		container.Init(s.ctx, logic.WithErrNotifyNeqCode(workerr.MysqlErrorCode, workerr.CacheErrorCode, workerr.RedisErrorCode))
	}

	return container.Run()
}

type PProfCollector struct {
	Action
}

func (s *PProfCollector) Desc() string {
	return "pprof采集"
}

func (s *PProfCollector) OnceKey() *OnceKey {
	return nil
}

func (s *PProfCollector) Exec() (err error) {
	// 创建采集器（每 5 分钟采集一次，禁用 CPU Profiling）
	collector := app.NewPProfCollector()

	// 启动采集任务
	collector.Start()
	return
}

type HeartBeatCollector struct {
	Action
}

func (s *HeartBeatCollector) Desc() string {
	return "pprof采集"
}

func (s *HeartBeatCollector) OnceKey() *OnceKey {
	return nil
}

func (s *HeartBeatCollector) Exec() (err error) {
	app.HeartBeat()
	return
}

type GlobalRewardPushNotify struct {
	Action
}

func (s *GlobalRewardPushNotify) Desc() string {
	return "全局奖品重新回调"
}

func (s *GlobalRewardPushNotify) OnceKey() *OnceKey {
	return &OnceKey{
		exp:     2 * time.Minute,
		key:     ckey.CronGlobalRewardPushNotify(),
		release: true,
	}
}

func (s *GlobalRewardPushNotify) Exec() (err error) {
	m := dao.RewardPushTask

	//找到需要重新回调的任务
	list, err := m.WithContext(s.ctx).Where(m.DoTime.Lte(time.Now().Add(-time.Hour).Unix())).Find()
	if err != nil {
		return
	}

	if len(list) == 0 {
		return
	}

	for _, v := range list {
		producer.GlobalRewardPushPub(s.ctx, 1, types.DelayCommonGlobalRewardPush{TaskId: v.ID})
	}

	return
}