package action import ( "joybuywork/internal/consts/ckey" "joybuywork/internal/repository/dao" "joybuywork/internal/repository/model" "joybuywork/internal/repository/mtype" "joybuywork/internal/service/provider/stocks" "joybuywork/internal/service/queue/producer" "joybuywork/internal/service/queue/types" "sync" "time" "github.com/panjf2000/ants/v2" "go.uber.org/atomic" "gorm.io/gen" ) type MallItemTask struct { Action } func (s *MallItemTask) Desc() string { return "商城奖品任务" } func (s *MallItemTask) OnceKey() *OnceKey { return &OnceKey{ exp: 2 * time.Minute, key: ckey.CronMallItemTask(), release: true, } } func (s *MallItemTask) LoggerMod() string { return loggerModDb } func (s *MallItemTask) DbExec() (total, successNum, failNum int64, err error) { m := dao.RedeemMallItemTask list, err := m.WithContext(s.ctx).Where(m.DoAt.Lte(time.Now().Unix())).Find() if err != nil { return } if len(list) == 0 { return } total = int64(len(list)) gp, _ := ants.NewPool(20) defer gp.Release() wg := sync.WaitGroup{} sn, fn := atomic.NewInt64(0), atomic.NewInt64(0) for _, v := range list { wg.Add(1) switch v.Mark { case mtype.RedeemMallItemTaskMarkDown: gp.Submit(func() { defer wg.Done() if err1 := s._down(v); err1 != nil { fn.Add(1) } else { sn.Add(1) } }) } } wg.Wait() successNum = sn.Load() failNum = fn.Load() return } func (s *MallItemTask) _down(task *model.RedeemMallItemTask) (err error) { m := dao.RedeemMallItem stock, err := stocks.NewMallItem(s.ctx, task.ItemID) if err != nil { return } useStock, err := stock.GetUseStock(task.ItemID) if err != nil { return } _, err = m.WithContext(s.ctx).Where(m.ID.Eq(task.ItemID)).UpdateSimple( m.UseStock.Value(useStock), m.DownAt.Value(task.DoAt), m.DownFrom.Value(mtype.RedeemMallItemDownFromTiming), ) if err != nil { return } _, err = dao.RedeemMallItemTask.WithContext(s.ctx).Delete(task) if err != nil { return } return } type ProjectUserCoinExpiredTask struct { Action } func (s *ProjectUserCoinExpiredTask) Desc() string { return "项目用户过期积分任务" } func (s *ProjectUserCoinExpiredTask) OnceKey() *OnceKey { return &OnceKey{ exp: 11 * time.Minute, key: ckey.CronProjectUserCoinExpiredTask(), release: true, } } func (s *ProjectUserCoinExpiredTask) LoggerMod() string { return loggerModDb } func (s *ProjectUserCoinExpiredTask) DbExec() (total, successNum, failNum int64, err error) { m := dao.ProjectUserCoinExpired var list []*model.ProjectUserCoinExpired err = m.WithContext(s.ctx). Select(m.ID). Where( m.DoneAt.Lte(time.Now().Add(10*time.Minute).Unix()), m.IsWait.Eq(mtype.No), ). FindInBatches(&list, 200, func(tx gen.Dao, batch int) (err error) { total += int64(len(list)) var ids []int64 for _, v := range list { ids = append(ids, v.ID) producer.CoinExpired(s.ctx, int(v.DoneAt-time.Now().Unix()), types.DelayCommonCoinExpired{ TaskId: v.ID, }) successNum++ } _, err = m.WithContext(s.ctx).Where(m.ID.In(ids...)).UpdateSimple(m.IsWait.Value(mtype.Yes)) return }) if err != nil { return } return } func (s *ProjectUserCoinExpiredTask) _down(task *model.RedeemMallItemTask) (err error) { m := dao.RedeemMallItem stock, err := stocks.NewMallItem(s.ctx, task.ItemID) if err != nil { return } useStock, err := stock.GetUseStock(task.ItemID) if err != nil { return } _, err = m.WithContext(s.ctx).Where(m.ID.Eq(task.ItemID)).UpdateSimple( m.UseStock.Value(useStock), m.DownAt.Value(task.DoAt), m.DownFrom.Value(mtype.RedeemMallItemDownFromTiming), ) if err != nil { return } _, err = dao.RedeemMallItemTask.WithContext(s.ctx).Delete(task) if err != nil { return } return }