liucaijun
/
voucher
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

预警9

This commit is contained in:
ziming 2025-07-01 18:12:03 +08:00
parent ee77d285fa
commit d57225b9a9
12 changed files with 211 additions and 202 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
internal/biz/alarm.go
View File

@ -9,12 +9,12 @@ import (
"voucher/internal/pkg/lock"
)
func (v *VoucherBiz) alarm(ctx context.Context, order *bo.OrderBo, errMsg string) error {
func (this *VoucherBiz) alarm(ctx context.Context, order *bo.OrderBo, errMsg string) error {
// 1小时 内 指定的批次号 发放 发生错误 预警
c := vo.OrderConsumeFailAlarmKey.BuildCache([]string{order.ProductNo})
_, err := v.rdb.Rdb.Get(ctx, c.Key).Result()
_, err := this.rdb.Rdb.Get(ctx, c.Key).Result()
if err == nil {
// 缓存存在，直接返回
@ -22,18 +22,18 @@ func (v *VoucherBiz) alarm(ctx context.Context, order *bo.OrderBo, errMsg string
}
if err != redis.Nil {
return fmt.Errorf(fmt.Sprintf("alarm 获取redis缓存%s异常:%v", c.Key, err))
return fmt.Errorf(fmt.Sprintf("alarm 获取redis缓存%s异常:%this", c.Key, err))
}
cl := vo.OrderConsumeFailAlarmLockKey.BuildCache([]string{order.ProductNo})
return lock.NewMutex(v.rdb.Rdb, cl.TTL).Lock(ctx, cl.Key, func(ctx context.Context) error {
return lock.NewMutex(this.rdb.Rdb, cl.TTL).Lock(ctx, cl.Key, func(ctx context.Context) error {
// 二次获取，判定处理，以免获取锁后又执行了一次
cacheValue, err3 := v.rdb.Rdb.Get(ctx, c.Key).Result()
cacheValue, err3 := this.rdb.Rdb.Get(ctx, c.Key).Result()
if err3 != nil && err3 != redis.Nil {
return fmt.Errorf(fmt.Sprintf("alarm 二次获取redis缓存%s异常:%v", c.Key, err))
return fmt.Errorf(fmt.Sprintf("alarm 二次获取redis缓存%s异常:%this", c.Key, err))
}
if len(cacheValue) > 0 {
@ -41,19 +41,19 @@ func (v *VoucherBiz) alarm(ctx context.Context, order *bo.OrderBo, errMsg string
}
// 通知
if err = v.DingMixRepo.SendMarkdownMessage(ctx, "异常通知", v.alarmText(ctx, order, errMsg)); err != nil {
if err = this.DingMixRepo.SendMarkdownMessage(ctx, "异常通知", this.alarmText(ctx, order, errMsg)); err != nil {
return err
}
if err = v.rdb.Rdb.Set(ctx, c.Key, order.ProductNo, c.TTL).Err(); err != nil {
return fmt.Errorf(fmt.Sprintf("设置redis缓存%s异常:%v", c.Key, err))
if err = this.rdb.Rdb.Set(ctx, c.Key, order.ProductNo, c.TTL).Err(); err != nil {
return fmt.Errorf(fmt.Sprintf("设置redis缓存%s异常:%this", c.Key, err))
}
return nil
})
}
func (v *VoucherBiz) alarmText(_ context.Context, order *bo.OrderBo, errMsg string) string {
func (this *VoucherBiz) alarmText(_ context.Context, order *bo.OrderBo, errMsg string) string {
remarks := fmt.Sprintf("订单号:%s，商品编号:%s,原因:%s", order.OrderNo, order.ProductNo, errMsg)

62
internal/biz/cron_notice.go
View File

@ -14,46 +14,46 @@ import (
"voucher/internal/pkg/lock"
)
func (v *VoucherBiz) isCanNotice(ctx context.Context) error {
func (this *VoucherBiz) isCanNotice(ctx context.Context) error {
if v.bc.Cmb.NoticeStartDays == 0 {
if this.bc.Cmb.NoticeStartDays == 0 {
return errors.New("订单定时通知,noticeStartDays eq 0")
}
if v.bc.Cmb.NoticeEndDays == 0 {
if this.bc.Cmb.NoticeEndDays == 0 {
return errors.New("订单定时通知,noticeEndDays eq 0")
}
cache := vo.CmbBatchNoticeCacheKey.BuildCache([]string{""})
_, err := v.rdb.Rdb.Get(ctx, cache.Key).Result()
_, err := this.rdb.Rdb.Get(ctx, cache.Key).Result()
if err == nil {
return fmt.Errorf("订单定时通知,notice 获取redis缓存存在，已被执行,本台服务不做执行")
}
if err != redis.Nil {
return fmt.Errorf(fmt.Sprintf("订单定时通知,notice 获取redis缓存%s异常:%v", cache.Key, err))
return fmt.Errorf(fmt.Sprintf("订单定时通知,notice 获取redis缓存%s异常:%this", cache.Key, err))
}
c := vo.CmbBatchNoticeLockKey.BuildCache([]string{""})
return lock.NewMutex(v.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
return lock.NewMutex(this.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
// 二次获取，判定处理，以免获取锁后又执行了一次
cacheValue, err2 := v.rdb.Rdb.Get(ctx, cache.Key).Result()
cacheValue, err2 := this.rdb.Rdb.Get(ctx, cache.Key).Result()
if err2 != nil && err2 != redis.Nil {
return fmt.Errorf(fmt.Sprintf("订单定时通知,notice 二次获取redis缓存%s异常:%v", cache.Key, err2))
return fmt.Errorf(fmt.Sprintf("订单定时通知,notice 二次获取redis缓存%s异常:%this", cache.Key, err2))
}
if len(cacheValue) > 0 {
return fmt.Errorf("订单定时通知,notice 二次获取redis缓存存在，已被执行,本台服务不做执行")
}
if err = v.rdb.Rdb.Set(ctx, cache.Key, fmt.Sprintf("%d_%d", v.bc.Cmb.NoticeStartDays, v.bc.Cmb.NoticeEndDays), c.TTL).Err(); err != nil {
return fmt.Errorf(fmt.Sprintf("notice 设置redis缓存%s异常:%v", cache.Key, err))
if err = this.rdb.Rdb.Set(ctx, cache.Key, fmt.Sprintf("%d_%d", this.bc.Cmb.NoticeStartDays, this.bc.Cmb.NoticeEndDays), c.TTL).Err(); err != nil {
return fmt.Errorf(fmt.Sprintf("notice 设置redis缓存%s异常:%this", cache.Key, err))
}
log.Warnf("订单定时通知,notice 获取redis缓存,不存在，开始处理")
@ -61,28 +61,28 @@ func (v *VoucherBiz) isCanNotice(ctx context.Context) error {
})
}
func (v *VoucherBiz) Notice(ctx context.Context) error {
func (this *VoucherBiz) Notice(ctx context.Context) error {
if err := v.isCanNotice(ctx); err != nil {
if err := this.isCanNotice(ctx); err != nil {
return err
}
now := time.Now()
// 获取 NoticeStartDays 天前的日期
noticeStartDay := now.AddDate(0, 0, int(-v.bc.Cmb.NoticeStartDays))
noticeStartDay := now.AddDate(0, 0, int(-this.bc.Cmb.NoticeStartDays))
// 获取 NoticeStartDays 天前 00:00:00 的时间
startTime := time.Date(noticeStartDay.Year(), noticeStartDay.Month(), noticeStartDay.Day(), 0, 0, 0, 0, noticeStartDay.Location())
// 获取 NoticeEndDays 天前的日期
noticeEndDay := now.AddDate(0, 0, int(-v.bc.Cmb.NoticeEndDays))
noticeEndDay := now.AddDate(0, 0, int(-this.bc.Cmb.NoticeEndDays))
// 获取 NoticeEndDays 天 23:59:59 的时间
endTime := time.Date(noticeEndDay.Year(), noticeEndDay.Month(), noticeEndDay.Day(), 23, 59, 59, 0, noticeEndDay.Location())
return v.timeSliceQuery(ctx, startTime, endTime)
return this.timeSliceQuery(ctx, startTime, endTime)
}
func (v *VoucherBiz) timeSliceQuery(ctx context.Context, startTime, endTime time.Time) error {
func (this *VoucherBiz) timeSliceQuery(ctx context.Context, startTime, endTime time.Time) error {
log.Warnf("订单定时通知,开始处理,按每两个小时分片处理,范围:%s到%s", startTime.Format(time.DateTime), endTime.Format(time.DateTime))
@ -110,18 +110,18 @@ func (v *VoucherBiz) timeSliceQuery(ctx context.Context, startTime, endTime time
defer func() {
if err := recover(); err != nil {
_, file, line, _ := runtime.Caller(1) // 1 表示获取当前调用者的调用信息
log.Errorf("订单定时通知,发生错误:req:%+v,err:%v,file:%s,line:%d", req, err, file, line)
log.Errorf("订单定时通知,发生错误:req:%+this,err:%this,file:%s,line:%d", req, err, file, line)
}
}()
return v.ExecuteNotice(ctx, req)
return this.ExecuteNotice(ctx, req)
})
}
return eg.Wait() // 仅返回第一个错误
}
func (v *VoucherBiz) ExecuteNotice(ctx context.Context, req *bo.FindInBatchesUseBo) error {
func (this *VoucherBiz) ExecuteNotice(ctx context.Context, req *bo.FindInBatchesUseBo) error {
start := time.Now()
@ -129,13 +129,13 @@ func (v *VoucherBiz) ExecuteNotice(ctx context.Context, req *bo.FindInBatchesUse
useNum := 0
sucNum := 0
err := v.OrderRepo.FindInBatches(ctx, req, func(ctx context.Context, rows []*bo.OrderBo) error {
err := this.OrderRepo.FindInBatches(ctx, req, func(ctx context.Context, rows []*bo.OrderBo) error {
for _, order := range rows {
num += 1
if err := v.notice(ctx, order, &useNum, &sucNum); err != nil {
log.Errorf("订单定时通知,err:%v", err)
if err := this.notice(ctx, order, &useNum, &sucNum); err != nil {
log.Errorf("订单定时通知,err:%this", err)
}
}
@ -150,15 +150,15 @@ func (v *VoucherBiz) ExecuteNotice(ctx context.Context, req *bo.FindInBatchesUse
"sucNum": sucNum, // 重置为成功数量
"elapsed": time.Now().Sub(start).String(),
}
log.Warnf("订单定时通知,%+v", logFields)
log.Warnf("订单定时通知,%+this", logFields)
return err
}
func (v *VoucherBiz) notice(ctx context.Context, order *bo.OrderBo, useNum, sucNum *int) (respErr error) {
func (this *VoucherBiz) notice(ctx context.Context, order *bo.OrderBo, useNum, sucNum *int) (respErr error) {
// 批量通知不做数据存储，量会很大
status, err := v.WechatCpnRepo.Query(ctx, order)
status, err := this.WechatCpnRepo.Query(ctx, order)
if err != nil {
return err
}
@ -182,11 +182,11 @@ func (v *VoucherBiz) notice(ctx context.Context, order *bo.OrderBo, useNum, sucN
Type: order.Type,
}
if err = v.request(ctx, order, notify); err != nil {
if err = this.request(ctx, order, notify); err != nil {
return err
}
if err = v.UpdateOrderStatus(ctx, order.ID, status); err != nil {
if err = this.UpdateOrderStatus(ctx, order.ID, status); err != nil {
return err
}
@ -199,7 +199,7 @@ func (v *VoucherBiz) notice(ctx context.Context, order *bo.OrderBo, useNum, sucN
return nil
}
func (v *VoucherBiz) request(ctx context.Context, order *bo.OrderBo, notify *bo.OrderNotifyBo) (respErr error) {
func (this *VoucherBiz) request(ctx context.Context, order *bo.OrderBo, notify *bo.OrderNotifyBo) (respErr error) {
defer func() {
if err := recover(); err != nil {
@ -208,7 +208,7 @@ func (v *VoucherBiz) request(ctx context.Context, order *bo.OrderBo, notify *bo.
stackSize := runtime.Stack(stackBuf, false)
// 获取调用栈信息
_, file, line, _ := runtime.Caller(1) // 1 表示获取当前调用者的调用信息
respErr = fmt.Errorf("request panic:%v, orderNo:%s, file:%s, line:%d, stack: %s", err, order.OrderNo, file, line, stackBuf[:stackSize])
respErr = fmt.Errorf("request panic:%this, orderNo:%s, file:%s, line:%d, stack: %s", err, order.OrderNo, file, line, stackBuf[:stackSize])
}
}()
@ -224,7 +224,7 @@ func (v *VoucherBiz) request(ctx context.Context, order *bo.OrderBo, notify *bo.
return nil // 不可通知，忽略
}
request, err := v.Cmb.NotifyRequest(ctx, order, notify)
request, err := this.Cmb.NotifyRequest(ctx, order, notify)
if err != nil {
return err
}
@ -233,7 +233,7 @@ func (v *VoucherBiz) request(ctx context.Context, order *bo.OrderBo, notify *bo.
return fmt.Errorf("request is nil,orderNo:%s,outBizNo:%s,stockId:%s,err:%s", order.OrderNo, order.OutBizNo, order.BatchNo)
}
if _, err = v.CmbMixRepo.Request(ctx, request, order.NotifyUrl); err != nil {
if _, err = this.CmbMixRepo.Request(ctx, request, order.NotifyUrl); err != nil {
return fmt.Errorf("orderNo:%s,outBizNo:%s,stockId:%s,err:%s", order.OrderNo, order.OutBizNo, order.BatchNo, err.Error())
}

24
internal/biz/notify_retry.go
View File

@ -12,25 +12,25 @@ import (
"voucher/internal/pkg/mq"
)
func (v *VoucherBiz) PushNotifyRetryDelayMQ(ctx context.Context, level int, orderNotifyId uint64) error {
func (this *VoucherBiz) PushNotifyRetryDelayMQ(ctx context.Context, level int, orderNotifyId uint64) error {
str := strconv.FormatUint(orderNotifyId, 10)
eventMap := v.bc.RocketMQ.EventMap["notifyRetry"]
eventMap := this.bc.RocketMQ.EventMap["notifyRetry"]
sendOption := []mq.SendOption{
mq.WithSendShardingKeysOption(str),
mq.WithOpenTelemetryOption(trace.SpanFromContext(ctx).SpanContext().TraceID().String()),
mq.WithSendDelayLevelOption(level),
}
if err := v.MqSendMixRepo.SendSync(ctx, eventMap.Topic, []byte("{}"), sendOption...); err != nil {
if err := this.MqSendMixRepo.SendSync(ctx, eventMap.Topic, []byte("{}"), sendOption...); err != nil {
return fmt.Errorf("回调通知延迟队列,投递消息出错：err=%s", err.Error())
}
return nil
}
func (v *VoucherBiz) NotifyRetryConsume(ctx context.Context, orderNotifyId uint64) error {
func (this *VoucherBiz) NotifyRetryConsume(ctx context.Context, orderNotifyId uint64) error {
var (
err error
@ -38,21 +38,21 @@ func (v *VoucherBiz) NotifyRetryConsume(ctx context.Context, orderNotifyId uint6
c = vo.NotifyRetryConsume.BuildCacheUint64([]uint64{orderNotifyId})
)
err = lock.NewMutex(v.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
err = lock.NewMutex(this.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
orderNotify, err = v.OrderNotifyRepo.GetByID(ctx, orderNotifyId)
orderNotify, err = this.OrderNotifyRepo.GetByID(ctx, orderNotifyId)
if err != nil {
return err
}
order, err2 := v.OrderRepo.GetByOrderNo(ctx, orderNotify.OrderNo)
order, err2 := this.OrderRepo.GetByOrderNo(ctx, orderNotify.OrderNo)
if err2 != nil {
return err2
}
if order.Type.IsCmb() {
orderNotify, err2 = v.Cmb.NotifyRetryConsume(ctx, order, orderNotify)
orderNotify, err2 = this.Cmb.NotifyRetryConsume(ctx, order, orderNotify)
return err2
}
@ -63,18 +63,18 @@ func (v *VoucherBiz) NotifyRetryConsume(ctx context.Context, orderNotifyId uint6
return err
}
level, err2 := v.level(ctx, orderNotify)
level, err2 := this.level(ctx, orderNotify)
if err2 != nil {
return err2
}
return v.PushNotifyRetryDelayMQ(ctx, level, orderNotify.ID)
return this.PushNotifyRetryDelayMQ(ctx, level, orderNotify.ID)
}
func (v *VoucherBiz) level(ctx context.Context, orderNotify *bo.OrderNotifyBo) (int, error) {
func (this *VoucherBiz) level(ctx context.Context, orderNotify *bo.OrderNotifyBo) (int, error) {
// 状态回调接口失败需要支持重试 重试间隔为1分钟、2分钟、12分钟、60分钟、360分钟
count, err := v.OrderNotifyRepo.GetCountByOrderNoAndEvent(ctx, orderNotify.OrderNo, orderNotify.Event)
count, err := this.OrderNotifyRepo.GetCountByOrderNoAndEvent(ctx, orderNotify.OrderNo, orderNotify.Event)
if err != nil {
return 0, err
}

60
internal/biz/order.go
View File

@ -8,9 +8,9 @@ import (
"voucher/internal/biz/vo"
)
func (v *VoucherBiz) CmbOrder(ctx context.Context, req *bo.OrderCreateReqBo) (orderNo string, err error) {
func (this *VoucherBiz) CmbOrder(ctx context.Context, req *bo.OrderCreateReqBo) (orderNo string, err error) {
order, err3 := v.GetByOutBizNo(ctx, req)
order, err3 := this.GetByOutBizNo(ctx, req)
if err3 != nil {
return "", err3
}
@ -19,7 +19,7 @@ func (v *VoucherBiz) CmbOrder(ctx context.Context, req *bo.OrderCreateReqBo) (or
if order.Status.IsFail() || order.Status.IsIng() {
if err4 := v.orderRetry(ctx, order); err4 != nil {
if err4 := this.orderRetry(ctx, order); err4 != nil {
return "", err4
}
}
@ -27,12 +27,12 @@ func (v *VoucherBiz) CmbOrder(ctx context.Context, req *bo.OrderCreateReqBo) (or
return order.OrderNo, err
}
product, err3 := v.ProductRepo.GetByProductNo(ctx, req.ProductNo)
product, err3 := this.ProductRepo.GetByProductNo(ctx, req.ProductNo)
if err3 != nil {
return "", err3
}
order, err3 = v.order(ctx, req, product)
order, err3 = this.order(ctx, req, product)
if err3 != nil {
return "", err3
}
@ -40,46 +40,46 @@ func (v *VoucherBiz) CmbOrder(ctx context.Context, req *bo.OrderCreateReqBo) (or
return order.OrderNo, nil
}
func (v *VoucherBiz) order(ctx context.Context, req *bo.OrderCreateReqBo, product *bo.ProductBo) (*bo.OrderBo, error) {
func (this *VoucherBiz) order(ctx context.Context, req *bo.OrderCreateReqBo, product *bo.ProductBo) (*bo.OrderBo, error) {
order, err := v.create(ctx, req, product)
order, err := this.create(ctx, req, product)
if err != nil {
return nil, err
}
voucherNo, err := v.WechatCpnRepo.Order(ctx, order)
voucherNo, err := this.WechatCpnRepo.Order(ctx, order)
if err != nil {
if err3 := v.fail(ctx, order, err); err3 != nil {
if err3 := this.fail(ctx, order, err); err3 != nil {
return nil, err3
}
return nil, err
}
if err = v.success(ctx, order, voucherNo); err != nil {
if err = this.success(ctx, order, voucherNo); err != nil {
return nil, err
}
return order, nil
}
func (v *VoucherBiz) orderRetry(ctx context.Context, order *bo.OrderBo) error {
func (this *VoucherBiz) orderRetry(ctx context.Context, order *bo.OrderBo) error {
voucherNo, err := v.WechatCpnRepo.Order(ctx, order)
voucherNo, err := this.WechatCpnRepo.Order(ctx, order)
if err != nil {
if err3 := v.fail(ctx, order, err); err3 != nil {
if err3 := this.fail(ctx, order, err); err3 != nil {
return err3
}
return err
}
return v.success(ctx, order, voucherNo)
return this.success(ctx, order, voucherNo)
}
func (v *VoucherBiz) create(ctx context.Context, req *bo.OrderCreateReqBo, product *bo.ProductBo) (*bo.OrderBo, error) {
func (this *VoucherBiz) create(ctx context.Context, req *bo.OrderCreateReqBo, product *bo.ProductBo) (*bo.OrderBo, error) {
o := &bo.OrderBo{
OrderNo: v.GenerateMixRepo.GeneratorString(ctx, fmt.Sprintf("%d%s", req.Type, req.OutBizNo)),
OrderNo: this.GenerateMixRepo.GeneratorString(ctx, fmt.Sprintf("%d%s", req.Type, req.OutBizNo)),
OutBizNo: req.OutBizNo,
ProductNo: req.ProductNo,
Account: req.Account,
@ -94,22 +94,22 @@ func (v *VoucherBiz) create(ctx context.Context, req *bo.OrderCreateReqBo, produ
Attach: req.Attach,
}
return v.OrderRepo.Create(ctx, o)
return this.OrderRepo.Create(ctx, o)
}
func (v *VoucherBiz) ing(ctx context.Context, id uint64) error {
func (this *VoucherBiz) ing(ctx context.Context, id uint64) error {
return v.OrderRepo.Ing(ctx, id)
return this.OrderRepo.Ing(ctx, id)
}
func (v *VoucherBiz) success(ctx context.Context, order *bo.OrderBo, voucherNo string) error {
func (this *VoucherBiz) success(ctx context.Context, order *bo.OrderBo, voucherNo string) error {
return v.OrderRepo.Success(ctx, order.ID, voucherNo)
return this.OrderRepo.Success(ctx, order.ID, voucherNo)
}
func (v *VoucherBiz) fail(ctx context.Context, order *bo.OrderBo, errReq error) error {
func (this *VoucherBiz) fail(ctx context.Context, order *bo.OrderBo, errReq error) error {
if err := v.OrderRepo.Fail(ctx, order.ID, errReq.Error()); err != nil {
if err := this.OrderRepo.Fail(ctx, order.ID, errReq.Error()); err != nil {
return err
}
@ -117,12 +117,12 @@ func (v *VoucherBiz) fail(ctx context.Context, order *bo.OrderBo, errReq error)
return nil // 过滤调该类型错误通知
}
return v.alarm(ctx, order, errReq.Error())
return this.alarm(ctx, order, errReq.Error())
}
func (v *VoucherBiz) GetByOutBizNo(ctx context.Context, req *bo.OrderCreateReqBo) (*bo.OrderBo, error) {
func (this *VoucherBiz) GetByOutBizNo(ctx context.Context, req *bo.OrderCreateReqBo) (*bo.OrderBo, error) {
order, err := v.OrderRepo.GetByOutBizNo(ctx, vo.OrderTypeCmb, req.OutBizNo)
order, err := this.OrderRepo.GetByOutBizNo(ctx, vo.OrderTypeCmb, req.OutBizNo)
if err != nil && !err2.IsDbNotFound(err) {
return nil, err
@ -131,19 +131,19 @@ func (v *VoucherBiz) GetByOutBizNo(ctx context.Context, req *bo.OrderCreateReqBo
return order, nil
}
func (v *VoucherBiz) UpdateOrderStatus(ctx context.Context, orderId uint64, status vo.OrderStatus) error {
func (this *VoucherBiz) UpdateOrderStatus(ctx context.Context, orderId uint64, status vo.OrderStatus) error {
if status.IsSuccess() {
return v.OrderRepo.Available(ctx, orderId)
return this.OrderRepo.Available(ctx, orderId)
} else if status.IsUse() {
return v.OrderRepo.Used(ctx, orderId)
return this.OrderRepo.Used(ctx, orderId)
} else if status.IsExpired() {
return v.OrderRepo.Expired(ctx, orderId)
return this.OrderRepo.Expired(ctx, orderId)
}
return fmt.Errorf("notice 未知券状态,orderId:%d,statuText:%s", orderId, status.GetText())

10
internal/biz/product.go
View File

@ -11,13 +11,13 @@ import (
"voucher/internal/pkg/lock"
)
func (v *VoucherBiz) CmbProductQuery(ctx context.Context, productNo string) (reps *v1.CmbQueryProductReply, err error) {
func (this *VoucherBiz) CmbProductQuery(ctx context.Context, productNo string) (reps *v1.CmbQueryProductReply, err error) {
c := vo.CmbProductQueryLockKey.BuildCache([]string{productNo})
err = lock.NewMutex(v.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
err = lock.NewMutex(this.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
product, err3 := v.ProductRepo.GetByProductNo(ctx, productNo)
product, err3 := this.ProductRepo.GetByProductNo(ctx, productNo)
if err3 != nil {
return err3
}
@ -26,7 +26,7 @@ func (v *VoucherBiz) CmbProductQuery(ctx context.Context, productNo string) (rep
return fmt.Errorf("只支持微信")
}
wechatResp, err4 := v.WechatCpnRepo.QueryProduct(ctx, product.MchId, product.BatchNo)
wechatResp, err4 := this.WechatCpnRepo.QueryProduct(ctx, product.MchId, product.BatchNo)
if err4 != nil {
return err4
}
@ -81,7 +81,7 @@ func (v *VoucherBiz) CmbProductQuery(ctx context.Context, productNo string) (rep
return
}
func (v *VoucherBiz) WxResp(wxResp *cashcoupons.Stock) (reps *do.WxResp) {
func (this *VoucherBiz) WxResp(wxResp *cashcoupons.Stock) (reps *do.WxResp) {
availableStock := *wxResp.StockUseRule.MaxCoupons - *wxResp.DistributedCoupons
couponAmount := *wxResp.StockUseRule.FixedNormalCoupon.CouponAmount / 100

22
internal/biz/query.go
View File

@ -11,18 +11,18 @@ import (
"voucher/internal/pkg/lock"
)
func (v *VoucherBiz) CmbQuery(ctx context.Context, orderNo string) (resp *v1.CmbQueryReply, err error) {
func (this *VoucherBiz) CmbQuery(ctx context.Context, orderNo string) (resp *v1.CmbQueryReply, err error) {
c := vo.CmbQueryLockKey.BuildCache([]string{orderNo})
err = lock.NewMutex(v.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
err = lock.NewMutex(this.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
order, err3 := v.OrderRepo.GetByOrderNo(ctx, orderNo)
order, err3 := this.OrderRepo.GetByOrderNo(ctx, orderNo)
if err3 != nil {
return err3
}
if err = v.Query(ctx, order); err != nil {
if err = this.Query(ctx, order); err != nil {
return err
}
@ -35,7 +35,7 @@ func (v *VoucherBiz) CmbQuery(ctx context.Context, orderNo string) (resp *v1.Cmb
Ticket: order.OrderNo,
Status: status.GetValue(),
TransDate: time.Now().Format("20060102150405"),
OrgNo: v.bc.Cmb.OrgNo,
OrgNo: this.bc.Cmb.OrgNo,
Ext: "",
}
@ -45,9 +45,9 @@ func (v *VoucherBiz) CmbQuery(ctx context.Context, orderNo string) (resp *v1.Cmb
return
}
func (v *VoucherBiz) Query(ctx context.Context, order *bo.OrderBo) error {
func (this *VoucherBiz) Query(ctx context.Context, order *bo.OrderBo) error {
status, err := v.WechatCpnRepo.Query(ctx, order)
status, err := this.WechatCpnRepo.Query(ctx, order)
if err != nil {
return err
}
@ -57,7 +57,7 @@ func (v *VoucherBiz) Query(ctx context.Context, order *bo.OrderBo) error {
return nil
}
if err = v.UpdateOrderStatus(ctx, order.ID, status); err != nil {
if err = this.UpdateOrderStatus(ctx, order.ID, status); err != nil {
return err
}
@ -66,14 +66,14 @@ func (v *VoucherBiz) Query(ctx context.Context, order *bo.OrderBo) error {
return nil
}
func (v *VoucherBiz) QueryOrder(ctx context.Context, orderNo string) (string, error) {
func (this *VoucherBiz) QueryOrder(ctx context.Context, orderNo string) (string, error) {
order, err3 := v.OrderRepo.GetByOrderNo(ctx, orderNo)
order, err3 := this.OrderRepo.GetByOrderNo(ctx, orderNo)
if err3 != nil {
return "", err3
}
status, err := v.WechatCpnRepo.Query(ctx, order)
status, err := this.WechatCpnRepo.Query(ctx, order)
if err != nil {
return "", err
}

24
internal/biz/register_tag.go
View File

@ -35,19 +35,19 @@ func (this *VoucherBiz) RegisterTag(ctx context.Context, batchNo string) error {
return err
}
func (v *VoucherBiz) registerNotifyTag(ctx context.Context, stockCreatorMchID, stockID string) error {
func (this *VoucherBiz) registerNotifyTag(ctx context.Context, stockCreatorMchID, stockID string) error {
cl := vo.WechatNotifyRegisterTagCacheLockKey.BuildCache([]string{v.bc.WechatNotifyMQ.Tag, stockCreatorMchID, stockID})
cl := vo.WechatNotifyRegisterTagCacheLockKey.BuildCache([]string{this.bc.WechatNotifyMQ.Tag, stockCreatorMchID, stockID})
return lock.NewMutex(v.rdb.Rdb, cl.TTL).Lock(ctx, cl.Key, func(ctx context.Context) error {
return lock.NewMutex(this.rdb.Rdb, cl.TTL).Lock(ctx, cl.Key, func(ctx context.Context) error {
wechatNotifyTag, err3 := v.WechatNotifyRegisterTagRepo.GetByStockIdAndMchId(ctx, stockCreatorMchID, stockID)
wechatNotifyTag, err3 := this.WechatNotifyRegisterTagRepo.GetByStockIdAndMchId(ctx, stockCreatorMchID, stockID)
if err3 != nil && !err2.IsDbNotFound(err3) {
return err3
}
if wechatNotifyTag != nil {
if wechatNotifyTag.Tag != v.bc.WechatNotifyMQ.Tag {
if wechatNotifyTag.Tag != this.bc.WechatNotifyMQ.Tag {
return fmt.Errorf("tag不一致，请检查tag配置:%s", wechatNotifyTag.Tag)
}
@ -55,24 +55,24 @@ func (v *VoucherBiz) registerNotifyTag(ctx context.Context, stockCreatorMchID, s
return nil
}
} else {
wechatNotifyTag, err3 = v.createWechatNotifyRegisterTag(ctx, stockCreatorMchID, stockID)
wechatNotifyTag, err3 = this.createWechatNotifyRegisterTag(ctx, stockCreatorMchID, stockID)
if err3 != nil {
return err3
}
}
if err := v.WechatCpnRepo.RegisterNotifyTag(ctx, stockID); err != nil {
return v.WechatNotifyRegisterTagRepo.Fail(ctx, wechatNotifyTag.ID, err.Error())
if err := this.WechatCpnRepo.RegisterNotifyTag(ctx, stockID); err != nil {
return this.WechatNotifyRegisterTagRepo.Fail(ctx, wechatNotifyTag.ID, err.Error())
}
return v.WechatNotifyRegisterTagRepo.Success(ctx, wechatNotifyTag.ID)
return this.WechatNotifyRegisterTagRepo.Success(ctx, wechatNotifyTag.ID)
})
}
func (v *VoucherBiz) createWechatNotifyRegisterTag(ctx context.Context, stockCreatorMchID, stockID string) (*bo.WechatNotifyRegisterTagBo, error) {
return v.WechatNotifyRegisterTagRepo.Create(ctx, &bo.WechatNotifyRegisterTagBo{
func (this *VoucherBiz) createWechatNotifyRegisterTag(ctx context.Context, stockCreatorMchID, stockID string) (*bo.WechatNotifyRegisterTagBo, error) {
return this.WechatNotifyRegisterTagRepo.Create(ctx, &bo.WechatNotifyRegisterTagBo{
StockID: stockID,
StockCreatorMchID: stockCreatorMchID,
Tag: v.bc.WechatNotifyMQ.Tag,
Tag: this.bc.WechatNotifyMQ.Tag,
})
}

12
internal/biz/retry.go
View File

@ -7,22 +7,22 @@ import (
"voucher/internal/biz/vo"
)
func (v *VoucherBiz) OrderRetry(ctx context.Context, outBizNos []string) error {
func (this *VoucherBiz) OrderRetry(ctx context.Context, outBizNos []string) error {
if len(outBizNos) > 0 {
for _, outBizNo := range outBizNos {
order, err := v.OrderRepo.GetByOutBizNo(ctx, vo.OrderTypeCmb, outBizNo)
order, err := this.OrderRepo.GetByOutBizNo(ctx, vo.OrderTypeCmb, outBizNo)
if err != nil {
return fmt.Errorf(fmt.Sprintf("获取订单%s异常:%v", outBizNo, err))
return fmt.Errorf(fmt.Sprintf("获取订单%s异常:%this", outBizNo, err))
}
if !order.Status.IsIng() {
return fmt.Errorf(fmt.Sprintf("订单%s状态异常:%s", order.OrderNo, order.Status))
}
if err4 := v.orderRetry(ctx, order); err4 != nil {
if err4 := this.orderRetry(ctx, order); err4 != nil {
return err4
}
}
@ -30,11 +30,11 @@ func (v *VoucherBiz) OrderRetry(ctx context.Context, outBizNos []string) error {
return nil
}
return v.OrderRepo.FindIngInBatches(ctx, func(ctx context.Context, rows []*bo.OrderBo) error {
return this.OrderRepo.FindIngInBatches(ctx, func(ctx context.Context, rows []*bo.OrderBo) error {
for _, order := range rows {
if err4 := v.orderRetry(ctx, order); err4 != nil {
if err4 := this.orderRetry(ctx, order); err4 != nil {
return err4
}
}

73
internal/biz/warning_budget.go
View File

@ -12,60 +12,69 @@ import (
"voucher/internal/biz/vo"
)
func (s *VoucherBiz) WarningBudgetIncrKv(uid string) (string, time.Duration) {
func (this *VoucherBiz) WarningBudgetIncrKv(uid string) (string, time.Duration) {
v := vo.WarningBudgetSendIncr.BuildCache([]string{uid})
return v.Key, v.TTL
}
func (s *VoucherBiz) WarningBudgetIncr(ctx context.Context, key string, TTL time.Duration) (int64, error) {
func (this *VoucherBiz) WarningBudgetIncr(ctx context.Context, key string, TTL time.Duration) (int64, error) {
// 增加发送计数
count, err := s.rdb.Rdb.IncrBy(ctx, key, 1).Result()
count, err := this.rdb.Rdb.IncrBy(ctx, key, 1).Result()
if err != nil {
return 0, err
}
// 如果是第一次发送，设置 过期时间
if count == 1 {
if err = s.rdb.Rdb.Expire(ctx, key, TTL).Err(); err != nil {
if err = this.rdb.Rdb.Expire(ctx, key, TTL).Err(); err != nil {
return 0, fmt.Errorf("设置过期时间失败: %v", err)
}
}
// 如果发送次数超过 “指定” 条，清除再来
if count > 24 { // 大约2小时
return 0, s.WarningBudgetIncrDel(ctx, key)
return 0, this.WarningBudgetIncrDel(ctx, key)
}
return count, nil
}
func (s *VoucherBiz) WarningBudgetIncrDel(ctx context.Context, key string) error {
func (this *VoucherBiz) WarningBudgetIncrDel(ctx context.Context, key string) error {
if _, err := s.rdb.Rdb.Del(ctx, key).Result(); err != nil {
//r, err := this.rdb.Rdb.Get(ctx, key).Result()
//if err != nil {
// return err
//}
//
//if len(r) == 0 {
// return nil
//}
if _, err := this.rdb.Rdb.Del(ctx, key).Result(); err != nil {
return err
}
return nil
}
func (v *VoucherBiz) WarningBudget(ctx context.Context) {
func (this *VoucherBiz) WarningBudget(ctx context.Context) {
uid := "warningBudget"
if b := v.Get(uid); b {
if b := this.Get(uid); b {
log.Warn("预警查询,上波还未执行完毕，此次暂不执行")
return
}
v.Add(uid)
defer v.Remove(uid)
this.Add(uid)
defer this.Remove(uid)
start := time.Now()
log.Warnf("预警查询,执行开始: %s", start.Format(time.DateTime))
if err := v.warningBudget(ctx); err != nil {
log.Errorf("预警查询,执行失败: %v", err)
if err := this.warningBudget(ctx); err != nil {
log.Errorf("预警查询,执行失败: %this", err)
}
end := time.Now()
@ -75,18 +84,18 @@ func (v *VoucherBiz) WarningBudget(ctx context.Context) {
return
}
func (v *VoucherBiz) warningBudget(ctx context.Context) error {
func (this *VoucherBiz) warningBudget(ctx context.Context) error {
err := v.ProductRepo.FindWarningBudget(ctx, func(ctx context.Context, rows []*bo.ProductBo) error {
err := this.ProductRepo.FindWarningBudget(ctx, func(ctx context.Context, rows []*bo.ProductBo) error {
for _, row := range rows {
wxResp, err := v.WechatCpnRepo.QueryProduct(ctx, row.MchId, row.BatchNo)
wxResp, err := this.WechatCpnRepo.QueryProduct(ctx, row.MchId, row.BatchNo)
if err != nil {
log.Context(ctx).Errorf("预警查询,查询微信券失败: %v", err)
log.Context(ctx).Errorf("预警查询,查询微信券失败: %this", err)
} else {
if err = v.Calculate(ctx, row, wxResp); err != nil {
log.Context(ctx).Errorf("预警查询,处理失败: %v", err)
if err = this.Calculate(ctx, row, wxResp); err != nil {
log.Context(ctx).Errorf("预警查询,处理失败: %this", err)
}
}
@ -103,42 +112,42 @@ func (v *VoucherBiz) warningBudget(ctx context.Context) error {
return nil
}
func (v *VoucherBiz) Calculate(ctx context.Context, product *bo.ProductBo, wxResp *cashcoupons.Stock) error {
func (this *VoucherBiz) Calculate(ctx context.Context, product *bo.ProductBo, wxResp *cashcoupons.Stock) error {
w := v.WxResp(wxResp)
w := this.WxResp(wxResp)
if err := v.ProductRepo.UpdateByWxResp(ctx, product.ID, w); err != nil {
if err := this.ProductRepo.UpdateByWxResp(ctx, product.ID, w); err != nil {
return err
}
k, t := this.WarningBudgetIncrKv(product.BatchNo)
if product.WarningBudget >= w.AvailableBudget {
k, t := v.WarningBudgetIncrKv(product.BatchNo)
count, err2 := v.WarningBudgetIncr(ctx, k, t)
count, err2 := this.WarningBudgetIncr(ctx, k, t)
if err2 != nil {
return err2
}
if count == 1 {
return v.WarningSend(ctx, product, w)
return this.WarningSend(ctx, product, w)
} else {
log.Warnf("预警查询,当前达到预警第[%d]次,暂不做通知", count)
}
}
if w.AllBudget > product.AllBudget {
if err := v.WarningBudgetIncrDel(ctx, k); err != nil {
return err
}
if w.AllBudget > product.AllBudget {
if err := this.WarningBudgetIncrDel(ctx, k); err != nil {
return err
}
}
return nil
}
func (v *VoucherBiz) WarningSend(ctx context.Context, product *bo.ProductBo, w *do.WxResp) error {
func (this *VoucherBiz) WarningSend(ctx context.Context, product *bo.ProductBo, w *do.WxResp) error {
err := v.DingMixRepo.SendMarkdownMessage(ctx, "券预算不足", formatAsCard(product, w))
err := this.DingMixRepo.SendMarkdownMessage(ctx, "券预算不足", formatAsCard(product, w))
if err != nil {
return err
}

42
internal/biz/wechat_notify.go
View File

@ -11,28 +11,28 @@ import (
"voucher/internal/pkg/lock"
)
func (v *VoucherBiz) WechatNotifyConsumer(ctx context.Context, tag string, req *bo.WechatVoucherNotifyBo) error {
func (this *VoucherBiz) WechatNotifyConsumer(ctx context.Context, tag string, req *bo.WechatVoucherNotifyBo) error {
c := vo.WechatNotifyConsumeLockKey.BuildCache([]string{tag, req.PlainText.StockID, req.PlainText.CouponID})
return lock.NewMutex(v.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
return lock.NewMutex(this.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
order, err := v.getOrder(ctx, req)
order, err := this.getOrder(ctx, req)
if err != nil {
return err
}
if req.PlainText.Status.IsSended() {
return v.available(ctx, order)
return this.available(ctx, order)
} else if req.PlainText.Status.IsUsed() {
return v.notifyUsed(ctx, order, req)
return this.notifyUsed(ctx, order, req)
} else if req.PlainText.Status.IsExpired() {
return v.expired(ctx, order)
return this.expired(ctx, order)
}
@ -73,62 +73,62 @@ func (this *VoucherBiz) getOrder(ctx context.Context, req *bo.WechatVoucherNotif
return order, nil
}
func (v *VoucherBiz) notifyUsed(ctx context.Context, order *bo.OrderBo, req *bo.WechatVoucherNotifyBo) error {
func (this *VoucherBiz) notifyUsed(ctx context.Context, order *bo.OrderBo, req *bo.WechatVoucherNotifyBo) error {
if order.Status.IsUse() {
// 券状态已是已使用，忽略不处理
return nil
}
if err := v.OrderRepo.NotifyUsed(ctx, order.ID, req.PlainText.ConsumeInformation.TransactionID); err != nil {
if err := this.OrderRepo.NotifyUsed(ctx, order.ID, req.PlainText.ConsumeInformation.TransactionID); err != nil {
return err
}
return v.notify(ctx, order)
return this.notify(ctx, order)
}
func (v *VoucherBiz) available(ctx context.Context, order *bo.OrderBo) error {
func (this *VoucherBiz) available(ctx context.Context, order *bo.OrderBo) error {
if order.Status.IsSuccess() {
// 券状态已是可使用，忽略不处理
return nil
}
if err := v.OrderRepo.Available(ctx, order.ID); err != nil {
if err := this.OrderRepo.Available(ctx, order.ID); err != nil {
return err
}
return v.notify(ctx, order)
return this.notify(ctx, order)
}
func (v *VoucherBiz) expired(ctx context.Context, order *bo.OrderBo) error {
func (this *VoucherBiz) expired(ctx context.Context, order *bo.OrderBo) error {
if order.Status.IsExpired() {
// 券状态已是已过期，忽略不处理
return nil
}
if err := v.OrderRepo.Expired(ctx, order.ID); err != nil {
if err := this.OrderRepo.Expired(ctx, order.ID); err != nil {
return err
}
//return v.notify(ctx, order)
//return this.notify(ctx, order)
return nil // 过期不做通知
}
func (v *VoucherBiz) notify(ctx context.Context, order *bo.OrderBo) error {
func (this *VoucherBiz) notify(ctx context.Context, order *bo.OrderBo) error {
return v.cmbNotify(ctx, order.ID)
return this.cmbNotify(ctx, order.ID)
}
func (v *VoucherBiz) cmbNotify(ctx context.Context, orderId uint64) error {
func (this *VoucherBiz) cmbNotify(ctx context.Context, orderId uint64) error {
order, err := v.OrderRepo.GetByID(ctx, orderId)
order, err := this.OrderRepo.GetByID(ctx, orderId)
if err != nil {
return err
}
if orderNotify, err := v.Cmb.Notify(ctx, order); err != nil {
if orderNotify, err := this.Cmb.Notify(ctx, order); err != nil {
if !errPb.IsNeedRetryNotify(err) {
return err
@ -136,7 +136,7 @@ func (v *VoucherBiz) cmbNotify(ctx context.Context, orderId uint64) error {
// 第一次通知失败重试入队
// 状态回调接口失败需要支持重试 重试间隔为1分钟、2分钟、12分钟、60分钟、360分钟
return v.PushNotifyRetryDelayMQ(ctx, 60, orderNotify.ID)
return this.PushNotifyRetryDelayMQ(ctx, 60, orderNotify.ID)
}
return nil

46
internal/biz/wechat_query.go
View File

@ -12,20 +12,20 @@ import (
"voucher/internal/biz/do"
)
func (v *VoucherBiz) uid(_ context.Context, msg string) string {
func (this *VoucherBiz) uid(_ context.Context, msg string) string {
return util.Md5(msg)
}
func (v *VoucherBiz) PushWechatQuery(ctx http.Context, req *do.WechatQuery) error {
func (this *VoucherBiz) PushWechatQuery(ctx http.Context, req *do.WechatQuery) error {
if req.ProductNo != "" {
_, err := v.ProductRepo.GetByProductNo(ctx, req.ProductNo)
_, err := this.ProductRepo.GetByProductNo(ctx, req.ProductNo)
if err != nil {
return err
}
}
queue := v.bc.RdsMQ.GetWechatQuery()
queue := this.bc.RdsMQ.GetWechatQuery()
if queue == nil {
return fmt.Errorf("队列不存在")
}
@ -37,25 +37,25 @@ func (v *VoucherBiz) PushWechatQuery(ctx http.Context, req *do.WechatQuery) erro
strMsg := string(msg)
uid := v.uid(ctx, strMsg)
if v.Get(uid) {
uid := this.uid(ctx, strMsg)
if this.Get(uid) {
return fmt.Errorf("此台服务队列正在处理中,key:%s,ip:%s", uid, ctx.Header().Get("X-Forwarded-For"))
}
v.Add(uid)
this.Add(uid)
_, err = v.rdb.Rdb.RPush(ctx, queue.Name, strMsg).Result()
_, err = this.rdb.Rdb.RPush(ctx, queue.Name, strMsg).Result()
if err != nil {
v.Remove(uid)
return fmt.Errorf("添加到队列失败:%v", err)
this.Remove(uid)
return fmt.Errorf("添加到队列失败:%this", err)
}
return nil
}
func (v *VoucherBiz) WechatQuery(ctx context.Context, msg string) error {
func (this *VoucherBiz) WechatQuery(ctx context.Context, msg string) error {
defer v.Remove(v.uid(ctx, msg))
defer this.Remove(this.uid(ctx, msg))
var req *do.WechatQuery
@ -72,14 +72,14 @@ func (v *VoucherBiz) WechatQuery(ctx context.Context, msg string) error {
n := 0
num := 0
notifyNum := 0
err := v.OrderRepo.FinSucByStockIdInBatches(ctx, req, func(ctx context.Context, rows []*bo.OrderBo) error {
err := this.OrderRepo.FinSucByStockIdInBatches(ctx, req, func(ctx context.Context, rows []*bo.OrderBo) error {
n += 1
for _, order := range rows {
num += 1
if err := v.wechatQuery(ctx, order, &notifyNum); err != nil {
log.Errorf("微信查询券订单状态发生错误,msg:%s,orderNo:%s,couponId:%s,appId:%s,openId:%s,stockId:%s,err:%v",
if err := this.wechatQuery(ctx, order, &notifyNum); err != nil {
log.Errorf("微信查询券订单状态发生错误,msg:%s,orderNo:%s,couponId:%s,appId:%s,openId:%s,stockId:%s,err:%this",
msg, order.OrderNo, order.VoucherNo, order.AppID, order.Account, order.BatchNo, err)
}
@ -98,33 +98,33 @@ func (v *VoucherBiz) WechatQuery(ctx context.Context, msg string) error {
return err
}
func (v *VoucherBiz) wechatQuery(ctx context.Context, order *bo.OrderBo, notifyNum *int) error {
func (this *VoucherBiz) wechatQuery(ctx context.Context, order *bo.OrderBo, notifyNum *int) error {
status, err := v.WechatCpnRepo.Query(ctx, order)
status, err := this.WechatCpnRepo.Query(ctx, order)
if err != nil {
return err
}
if status.IsUse() {
return v.queryUsed(ctx, order, notifyNum)
return this.queryUsed(ctx, order, notifyNum)
} else if status.IsExpired() {
return v.expired(ctx, order)
return this.expired(ctx, order)
}
return nil
}
func (v *VoucherBiz) queryUsed(ctx context.Context, order *bo.OrderBo, notifyNum *int) error {
func (this *VoucherBiz) queryUsed(ctx context.Context, order *bo.OrderBo, notifyNum *int) error {
*notifyNum += 1
if order.Status.IsUse() {
return v.notify(ctx, order)
return this.notify(ctx, order)
}
if err := v.OrderRepo.Used(ctx, order.ID); err != nil {
if err := this.OrderRepo.Used(ctx, order.ID); err != nil {
return err
}
return v.notify(ctx, order)
return this.notify(ctx, order)
}

18
internal/biz/wechat_retry.go
View File

@ -8,34 +8,34 @@ import (
"voucher/internal/biz/bo"
)
func (v *VoucherBiz) PushWechatRetry(ctx context.Context, batchNo string) error {
func (this *VoucherBiz) PushWechatRetry(ctx context.Context, batchNo string) error {
product, err := v.ProductRepo.GetByBatchNo(ctx, batchNo)
product, err := this.ProductRepo.GetByBatchNo(ctx, batchNo)
if err != nil {
return err
}
queue := v.bc.RdsMQ.GetWechatRetry()
queue := this.bc.RdsMQ.GetWechatRetry()
if queue == nil {
return fmt.Errorf("队列不存在")
}
_, err = v.rdb.Rdb.RPush(ctx, queue.Name, product.BatchNo).Result()
_, err = this.rdb.Rdb.RPush(ctx, queue.Name, product.BatchNo).Result()
if err != nil {
return fmt.Errorf("添加到队列失败:%v", err)
return fmt.Errorf("添加到队列失败:%this", err)
}
return nil
}
func (v *VoucherBiz) WechatRetry(ctx context.Context, batchNo string) error {
func (this *VoucherBiz) WechatRetry(ctx context.Context, batchNo string) error {
start := time.Now()
log.Warnf("失败订单重试开始:%s,batchNo:%s", start.String(), batchNo)
fmt.Printf("失败订单重试开始:%s,batchNo:%s", start.String(), batchNo)
num := 0
err := v.OrderRepo.FinFailByStockIdInBatches(ctx, batchNo, func(ctx context.Context, rows []*bo.OrderBo) error {
err := this.OrderRepo.FinFailByStockIdInBatches(ctx, batchNo, func(ctx context.Context, rows []*bo.OrderBo) error {
if len(rows) == 0 {
log.Infof("微信查询券订单状态,batchNo[%s],已处理[%d]单，无订单,结束执行", batchNo, num)
@ -45,8 +45,8 @@ func (v *VoucherBiz) WechatRetry(ctx context.Context, batchNo string) error {
for _, order := range rows {
num += 1
if err := v.orderRetry(ctx, order); err != nil {
log.Errorf("失败订单重试发生错误,batchNo:%s,orderNo:%s,appId:%s,openId:%s,err:%v",
if err := this.orderRetry(ctx, order); err != nil {
log.Errorf("失败订单重试发生错误,batchNo:%s,orderNo:%s,appId:%s,openId:%s,err:%this",
batchNo, order.OrderNo, order.AppID, order.Account, err)
}