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

package action

import (
	"joybuywork/internal/app"
	"joybuywork/internal/consts"
	"joybuywork/internal/consts/ckey"
	"joybuywork/internal/repository/cache"
	"joybuywork/internal/repository/dao"
	"joybuywork/internal/repository/model"
	"joybuywork/internal/repository/mtype"
	"joybuywork/internal/repository/withdefault"
	"joybuywork/internal/service/cmm"
	"joybuywork/internal/service/crons/action/containers"
	"joybuywork/internal/service/provider/cachecurd"
	"joybuywork/internal/service/provider/stocks"
	"joybuywork/internal/service/queue/producer"
	"joybuywork/internal/service/queue/types"
	"joybuywork/internal/utils/logic"
	"joybuywork/internal/utils/timez"
	"joybuywork/proto/go_proto/workerr"
	"fmt"
	"sync"
	"time"

	"github.com/golang-module/carbon/v2"
	"github.com/panjf2000/ants/v2"
)

type CampaignStart struct {
	Action
}

func (s *CampaignStart) Desc() string {
	return "活动开启"
}

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

func (s *CampaignStart) Exec() (err error) {
	m := dao.FissionCampaign

	list, err := m.WithContext(s.ctx).Where(
		m.StartTime.Gt(0),
		m.StartTime.Lte(time.Now().Unix()),
		m.Status.Eq(mtype.FissionCampaignStatusWait),
		m.IsDel.Eq(mtype.No),
	).Find()
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	if len(list) == 0 {
		return
	}

	var cpnMap = make(map[int64]*model.FissionCampaign, len(list))
	var ids, mapIds []int64
	for _, v := range list {
		ids = append(ids, v.ID)
		cpnMap[v.ID] = v
		if v.FissionID == consts.IdDice {
			mapIds = append(mapIds, v.ID)
		}
	}

	confM := dao.FissionCampaignConfig
	confs, err := confM.WithContext(s.ctx).Where(confM.CampaignID.In(ids...)).Find()
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	var pIds, ptIds []int64
	for _, v := range confs {
		if v.HasWhite == mtype.Yes {
			ptIds = append(ptIds, v.CampaignID)
			continue
		}
		pIds = append(pIds, v.CampaignID)
	}

	if len(pIds) > 0 {
		_, err = m.WithContext(s.ctx).Where(m.ID.In(pIds...)).UpdateSimple(m.Status.Value(mtype.FissionCampaignStatusPub))
		if err != nil {
			err = s.BE().MysqlError(err)
			return
		}

		unix := time.Now().Unix()
		dn := producer.NewDeepworkNotify(s.ctx)
		for _, id := range pIds {
			cpn := cpnMap[id]
			dn.CampaignStart(types.DnCampaignStart{
				CampaignId:     cpn.ID,
				CampaignHashId: cpn.HashID,
				Timestamp:      unix,
			})
		}
	}

	if len(ptIds) > 0 {
		_, err = m.WithContext(s.ctx).Where(m.ID.In(ptIds...)).UpdateSimple(m.Status.Value(mtype.FissionCampaignStatusPubPre))
		if err != nil {
			err = s.BE().MysqlError(err)
			return
		}
	}

	_ = cache.NewCampaign(s.ctx).Clear(list...)

	for _, id := range mapIds {
		producer.NewCampaignNotice(s.ctx).CampaignDiceMap(types.CampaignDiceMap{CampaignId: id})
	}

	return
}

type CampaignEnd struct {
	Action
}

func (s *CampaignEnd) Desc() string {
	return "活动结束"
}

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

func (s *CampaignEnd) Exec() (err error) {
	m := dao.FissionCampaign

	list, err := m.WithContext(s.ctx).Where(
		m.EndTime.Gt(0),
		m.EndTime.Lte(time.Now().Unix()),
		m.Status.In(mtype.FissionCampaignStatusPub, mtype.FissionCampaignStatusPubPre),
		m.IsDel.Eq(mtype.No),
	).Find()
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	var ids []int64
	for _, v := range list {
		ids = append(ids, v.ID)
	}

	if len(ids) == 0 {
		return
	}

	_, err = m.WithContext(s.ctx).Where(m.ID.In(ids...)).UpdateSimple(m.Status.Value(mtype.FissionCampaignStatusCancel))
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	_ = cache.NewCampaign(s.ctx).Clear(list...)

	unix := time.Now().Unix()
	dn := producer.NewDeepworkNotify(s.ctx)
	for _, cpn := range list {
		if cpn.ChildType == mtype.FissionCampaignChildTypeInvite {
			producer.NewCampaignNotice(s.ctx).CampaignChildInvite(types.CampaignChildInvite{CampaignId: cpn.ID})
		}

		err = cmm.NewCMM(s.ctx, dao.Q).CampaignCancelTask(cpn, false)
		if err != nil {
			return
		}

		dn.CampaignEnd(types.DnCampaignEnd{
			CampaignId:     cpn.ID,
			CampaignHashId: cpn.HashID,
			Timestamp:      unix,
		})
	}

	return
}

type CampaignPreToTun struct {
	Action
}

func (s *CampaignPreToTun) Desc() string {
	return "预发布转正式发布"
}

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

func (s *CampaignPreToTun) Exec() (err error) {
	m := dao.FissionCampaign

	list, err := m.WithContext(s.ctx).Where(
		m.PreToRun.Gt(0),
		m.PreToRun.Lte(time.Now().Unix()),
		m.Status.Eq(mtype.FissionCampaignStatusPubPre),
		m.IsDel.Eq(mtype.No),
	).Find()
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	if len(list) == 0 {
		return
	}

	var ids []int64
	for _, v := range list {
		ids = append(ids, v.ID)
	}

	confM := dao.FissionCampaignConfig
	confs, err := confM.WithContext(s.ctx).Where(confM.CampaignID.In(ids...)).Find()
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	var confIds []int64
	for _, v := range confs {
		confIds = append(confIds, v.ID)
	}

	_, err = m.WithContext(s.ctx).Where(m.ID.In(ids...)).UpdateSimple(m.Status.Value(mtype.FissionCampaignStatusPub))
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	_, err = confM.WithContext(s.ctx).Where(confM.ID.In(confIds...)).UpdateSimple(confM.HasWhite.Value(mtype.No))
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	unix := time.Now().Unix()
	dn := producer.NewDeepworkNotify(s.ctx)
	for _, cpn := range list {
		dn.CampaignStart(types.DnCampaignStart{
			CampaignId:     cpn.ID,
			CampaignHashId: cpn.HashID,
			Timestamp:      unix,
		})
	}

	_ = cache.NewCampaign(s.ctx).Clear(list...)

	_ = cache.NewCampaignConfig(s.ctx).Clear(confs...)

	return
}

type CampaignQuizOpen struct {
	Action
}

func (s *CampaignQuizOpen) Desc() string {
	return "有奖竞猜开奖"
}

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

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

	return container.Run()
}

type CampaignCompensateReward struct {
	Action
}

func (s *CampaignCompensateReward) Desc() string {
	return "活动奖品补偿"
}

func (s *CampaignCompensateReward) OnceKey() *OnceKey {
	return &OnceKey{
		exp:     5 * time.Second,
		key:     ckey.CronCampaignCompensateReward(),
		release: true,
	}
}

func (s *CampaignCompensateReward) Exec() (err error) {
	now := carbon.Now()

	start, end := now.SubHour(), now.SubMinutes(30)

	var ids []int64
	m := dao.FissionCampaign
	err = m.WithContext(s.ctx).Where(m.EndTime.Gte(end.Timestamp())).Pluck(m.ID, &ids)
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	wg := sync.WaitGroup{}
	defer wg.Wait()

	gp, _ := ants.NewPool(10)
	defer gp.Release()

	for _, id := range ids {
		wg.Add(1)
		cpnId := id

		_ = gp.Submit(func() {
			defer wg.Done()
			defer app.TryCatch(s.ctx, fmt.Sprintf("活动奖品补偿失败：cpnid：%d", cpnId))
			recordM := dao.FissionRewardRecord.SubTable(cpnId)

			list, err1 := recordM.WithContext(s.ctx).Where(
				recordM.CampaignID.Eq(cpnId),
				recordM.CreatedAt.Gte(start.Timestamp()),
				recordM.CreatedAt.Lte(end.Timestamp()),
				recordM.Status.Eq(mtype.FissionRewardRecordStatusWait),
			).Find()
			if err1 != nil {
				return
			}

			if len(list) == 0 {
				return
			}

			var recordMap = make(map[int64]*model.FissionRewardRecord)
			var recordIds []int64
			for _, v := range list {
				recordIds = append(recordIds, v.ID)
				recordMap[v.ID] = v
			}

			ntfM := dao.FissionNotify
			ntfList, err1 := ntfM.WithContext(s.ctx).Where(ntfM.CampaignID.Eq(cpnId), ntfM.RecordID.In(recordIds...)).Find()
			if err1 != nil {
				return
			}

			var ntfMap = make(map[int64]*model.FissionNotify)
			for _, v := range ntfList {
				ntfMap[v.RecordID] = v
				delete(recordMap, v.RecordID)
			}

			var sysErrIds []int64
			var okIds []int64
			var rewardErrNum int
			for _, v := range list {
				ntf, ok := ntfMap[v.ID]
				if !ok {
					producer.NewCampaignNotice(s.ctx).CampaignRewardNotify(types.CampaignRewardNotify{
						CampaignId:   cpnId,
						UserId:       v.UserID,
						UserRewardId: v.ID,
					})
					rewardErrNum++
					continue
				}

				if ntf.IsOk == mtype.No && ntf.PushNum >= mtype.FissionNotifyPushNumMax {
					sysErrIds = append(sysErrIds, v.ID)
					rewardErrNum++
					continue
				}

				if ntf.IsOk == mtype.Yes {
					okIds = append(okIds, v.ID)
					rewardErrNum++
					continue
				}

			}

			if len(sysErrIds) > 0 {
				_, err = recordM.WithContext(s.ctx).Where(recordM.ID.In(sysErrIds...)).UpdateSimple(
					recordM.Status.Value(mtype.FissionRewardRecordStatusFail),
					recordM.FailType.Value(mtype.FissionRewardRecordFailTypeSys),
				)
			}

			if len(okIds) > 0 {
				_, err = recordM.WithContext(s.ctx).Where(recordM.ID.In(okIds...)).UpdateSimple(
					recordM.Status.Value(mtype.FissionRewardRecordStatusSuccess),
				)
			}

			for _, record := range recordMap {
				producer.NewCampaignNotice(s.ctx).CampaignRewardNotify(types.CampaignRewardNotify{
					CampaignId:   record.CampaignID,
					UserId:       record.UserID,
					UserRewardId: record.ID,
				})
			}

			app.FeishuNormalSend(fmt.Sprintf("回调奖品丢失回调数据条数：%d，活动id：%d", rewardErrNum, cpnId))
		})
	}

	return
}

type CampaignCancelReward struct {
	Action
}

func (s *CampaignCancelReward) Desc() string {
	return "把养狗过期的奖品放回去奖品池"
}

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

func (s *CampaignCancelReward) Exec() (err error) {
	m := dao.FissionCampaign

	list, err := m.WithContext(s.ctx).Where(
		m.StartTime.Gt(0),
		m.StartTime.Lte(time.Now().Unix()),
		m.Status.In(mtype.FissionCampaignStatusPub, mtype.FissionCampaignStatusPubPre),
		m.IsDel.Eq(mtype.No),
		m.FissionID.Eq(consts.IdKeep),
	).Find()
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	if len(list) == 0 {
		return
	}

	wg := sync.WaitGroup{}
	defer wg.Wait()

	gp, _ := ants.NewPool(10)
	defer gp.Release()

	for _, v := range list {
		wg.Add(1)
		_ = gp.Submit(func() {
			defer wg.Done()
			defer app.TryCatch(s.ctx, fmt.Sprintf("把养狗过期的奖品放回去奖品池：cpnid：%d", v.ID))

			keepM := dao.FissionKeep
			exp, err1 := keepM.WithContext(s.ctx).
				Select(keepM.ID, keepM.RewardID).
				Where(
					keepM.CampaignID.Eq(v.ID),
					keepM.ExpiredAt.Lte(time.Now().Unix()),
					keepM.CancelReward.Eq(mtype.No),
					keepM.RewardID.Gt(0),
					keepM.RecordID.Eq(0),
				).Find()
			if err1 != nil {
				return
			}

			var cancelMap = make(map[int64][]int64)
			for _, keep := range exp {
				cancelMap[keep.RewardID] = append(cancelMap[keep.RewardID], keep.ID)
			}

			for rewardId, ids := range cancelMap {
				var stock *stocks.CampaignRewardSelect
				stock, err = stocks.NewCampaignRewardSelect(s.ctx, rewardId)
				if err != nil {
					return
				}

				stock.ClearUseStock()

				_, err = keepM.WithContext(s.ctx).Where(keepM.ID.In(ids...)).UpdateSimple(keepM.CancelReward.Value(mtype.Yes))
				if err != nil {
					return
				}
			}

		})
	}

	return
}

type CampaignCoinScene struct {
	Action
}

func (s *CampaignCoinScene) Desc() string {
	return "创建下一场金币池"
}

func (s *CampaignCoinScene) OnceKey() *OnceKey {
	return &OnceKey{
		exp:     21 * time.Minute,
		key:     ckey.CronCampaignCoinScene(),
		release: true,
	}
}

func (s *CampaignCoinScene) Exec() (err error) {
	m := dao.FissionCampaign

	list, err := m.WithContext(s.ctx).Where(
		m.StartTime.Gt(0),
		m.StartTime.Lte(time.Now().Unix()),
		m.Status.In(mtype.FissionCampaignStatusPub, mtype.FissionCampaignStatusPubPre),
		m.IsDel.Eq(mtype.No),
		m.FissionID.Eq(consts.IdCoin),
	).Find()
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	if len(list) == 0 {
		return
	}

	wg := sync.WaitGroup{}
	defer wg.Wait()

	gp, _ := ants.NewPool(10)
	defer gp.Release()

	for _, v := range list {
		wg.Add(1)
		_ = gp.Submit(func() {
			defer wg.Done()
			defer app.TryCatch(s.ctx, fmt.Sprintf("创建下一场金币池：cpnid：%d", v.ID))

			conf, _ := cache.NewCampaignCoin(s.ctx).FindByCampaignId(v.ID)

			sceneM := dao.FissionCoinScene

			if conf.OpenLimit > 0 { // 限制场次
				count, _ := sceneM.WithContext(s.ctx).Where(sceneM.CampaignID.Eq(v.ID)).Count()
				if conf.OpenLimit <= int32(count) {
					return
				}
			}

			_, start, _ := timez.GMTStartAndEndTs(v.TOffset)

			openTime := start + consts.DaySec // 明天开奖的场次

			nextOpenTime := openTime + int64(consts.DaySec*conf.OpenDays) // 下一场开奖时间
			nextRewardTime := nextOpenTime + consts.DaySec                // 下一场领奖结束时间

			if v.EndTime < nextOpenTime { // 活动结束不能生成
				return
			}

			// 看看明天有没有开奖的的，如果有那就要创建新的一场备用
			exit, _ := sceneM.WithContext(s.ctx).Where(sceneM.CampaignID.Eq(v.ID), sceneM.OpenTime.Eq(openTime)).Take()
			if exit == nil {
				return
			}

			// 看看是不是已经创建过了
			done, _ := sceneM.WithContext(s.ctx).Where(sceneM.CampaignID.Eq(v.ID), sceneM.OpenTime.Eq(nextOpenTime)).Take()
			if done != nil {
				return
			}

			err = sceneM.WithContext(s.ctx).Create(withdefault.FissionCoinScene(&model.FissionCoinScene{
				CampaignID: v.ID,
				OpenTime:   nextOpenTime,
				InitCoin:   conf.InitCoin,
				InitUse:    conf.InitUse,
				RewardTime: nextRewardTime,
				PreScene:   exit.ID,
			}))
			if err != nil {
				err = s.BE().MysqlError(err)
				return
			}

		})
	}

	return
}

type CoinSceneSettlement struct {
	Action
}

func (s *CoinSceneSettlement) Desc() string {
	return "金币场次结算"
}

func (s *CoinSceneSettlement) OnceKey() *OnceKey {
	return &OnceKey{
		exp:     21 * time.Minute,
		key:     ckey.CronCampaignCoinSceneSettlement(),
		release: true,
	}
}

func (s *CoinSceneSettlement) Exec() (err error) {
	now := time.Now().Truncate(time.Hour)

	// 就怕有错误时间差距
	start, end := now.Add(-5*time.Minute).Unix(), now.Add(5*time.Minute).Unix()

	m := dao.FissionCoinScene
	list, err := m.WithContext(s.ctx).Where(
		m.RewardTime.Between(start, end),
		m.IsArchive.Eq(mtype.No),
	).Find()
	if err != nil {
		err = s.BE().MysqlError(err)
		return
	}

	if len(list) == 0 {
		return
	}

	wg := sync.WaitGroup{}
	defer wg.Wait()

	gp, _ := ants.NewPool(10)
	defer gp.Release()

	for _, v := range list {
		wg.Add(1)
		_ = gp.Submit(func() {
			var err1 error
			defer func() {
				if err1 != nil {
					app.Logger.Error(s.ctx, fmt.Sprintf("场次处理失败：%v", err1))
				}
			}()

			defer wg.Done()

			key := ckey.FissionCoinSceneData(v.ID)
			scene, err1 := cachecurd.NewHMap(s.ctx, key,
				cachecurd.SetCoinSceneBuild(v.CampaignID, v.ID),
				cachecurd.SetHMapExp(consts.TimeDay),
			)
			if err1 != nil {
				return
			}

			joinNum, _ := scene.GetInt64(cachecurd.CoinSceneJoinNum)
			joinUserNum, _ := scene.GetInt64(cachecurd.CoinSceneJoinUserNum)
			userCoin, _ := scene.GetInt64(cachecurd.CoinSceneUserCoin)
			getNum, _ := scene.GetInt64(cachecurd.CoinSceneGetNum)
			getCoin, _ := scene.GetInt64(cachecurd.CoinSceneGetTotal)

			sceneM := dao.FissionCoinScene
			_, err1 = sceneM.WithContext(s.ctx).Where(sceneM.ID.Eq(v.ID)).UpdateSimple(
				sceneM.JoinNum.Value(joinNum),
				sceneM.JoinUserNum.Value(joinUserNum),
				sceneM.UserCoin.Value(userCoin),
				sceneM.GetNum.Value(getNum),
				sceneM.GetCoin.Value(getCoin),
				sceneM.IsArchive.Value(mtype.Yes),
			)
			if err1 != nil { // 有错误，没有人参与，参与人数等于领取人数
				return
			}

		})
	}

	return
}

// CampaignRewardAdd 按配置周期追加活动奖品库存。
type CampaignRewardAdd struct {
	Action
}

func (s *CampaignRewardAdd) Desc() string {
	return "活动奖品追加库存"
}

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

func (s *CampaignRewardAdd) Exec() (err error) {
	now := time.Now().Unix()
	m := dao.FissionCampaignRewardAdd
	list, err := m.WithContext(s.ctx).Where(
		m.Start.Lte(now),
		m.FinishAt.Eq(0),
		m.IsDel.Eq(mtype.No),
	).Find()
	if err != nil {
		return s.BE().MysqlError(err)
	}

	if len(list) == 0 {
		return
	}

	campaignMap := make(map[int64][]*model.FissionCampaignRewardAdd)
	for _, one := range list {
		campaignMap[one.CampaignID] = append(campaignMap[one.CampaignID], one)
	}

	gp, err := ants.NewPool(5)
	if err != nil {
		return err
	}
	defer gp.Release()

	var wg sync.WaitGroup
	var firstErr error
	for campaignID, rules := range campaignMap {
		wg.Add(1)
		_ = gp.Submit(func() {
			defer wg.Done()
			_ = s.addCampaignRewardStock(campaignID, rules, now)
		})
	}
	wg.Wait()

	return firstErr
}

func (s *CampaignRewardAdd) addCampaignRewardStock(campaignID int64, rules []*model.FissionCampaignRewardAdd, now int64) (err error) {
	var clearRewardIDs []int64
	err = dao.Q.Transaction(func(tx *dao.Query) error {
		addM := tx.FissionCampaignRewardAdd
		logM := tx.FissionCampaignRewardAddLog
		rewardM := tx.FissionCampaignReward

		for _, rule := range rules {
			interval := int64(rule.DayNum) * consts.DaySec
			calcAt := now
			finished := rule.End > 0 && now >= rule.End
			if finished {
				calcAt = rule.End
			}

			addAt := rule.Start + (calcAt-rule.Start)/interval*interval
			if addAt < rule.Start || (rule.End > 0 && addAt > rule.End) {
				continue
			}

			count, err1 := logM.WithContext(s.ctx).Where(
				logM.RewardID.Eq(rule.RewardID),
				logM.AddAt.Eq(addAt),
			).Count()
			if err1 != nil {
				return err1
			}

			if count == 0 {
				reward, err2 := rewardM.WithContext(s.ctx).Where(
					rewardM.ID.Eq(rule.RewardID),
					rewardM.CampaignID.Eq(campaignID),
				).Take()
				if err2 != nil {
					return err2
				}

				_, err2 = rewardM.WithContext(s.ctx).Where(rewardM.ID.Eq(reward.ID)).UpdateSimple(rewardM.Stock.Add(rule.DayAdd))
				if err2 != nil {
					return err2
				}

				err2 = logM.WithContext(s.ctx).Create(withdefault.FissionCampaignRewardAddLog(&model.FissionCampaignRewardAddLog{
					CampaignID: campaignID,
					RewardID:   rule.RewardID,
					AddAt:      addAt,
					DayAdd:     rule.DayAdd,
					BeforeAdd:  reward.Stock,
				}))
				if err2 != nil {
					return err2
				}
				clearRewardIDs = append(clearRewardIDs, rule.RewardID)
			}

			if finished {
				if _, err2 := addM.WithContext(s.ctx).Where(
					addM.ID.Eq(rule.ID),
					addM.FinishAt.Eq(0),
				).UpdateSimple(addM.FinishAt.Value(now)); err2 != nil {
					return err2
				}
			}
		}

		return nil
	})
	if err != nil {
		return s.BE().MysqlError(err)
	}

	stocks.NewCleaner(s.ctx, clearRewardIDs...).ClearAllReward()

	return
}