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 }