From b523b035ad812dcda81fc509fcb211894f7336c7 Mon Sep 17 00:00:00 2001 From: ziming Date: Wed, 28 May 2025 19:22:17 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E8=AF=95=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/biz/alarm.go | 69 +++++++++++++ internal/biz/order.go | 191 +---------------------------------- internal/biz/query.go | 44 ++++++++ internal/biz/register_tag.go | 102 +++++++++++++++++++ 4 files changed, 217 insertions(+), 189 deletions(-) create mode 100644 internal/biz/alarm.go create mode 100644 internal/biz/query.go create mode 100644 internal/biz/register_tag.go diff --git a/internal/biz/alarm.go b/internal/biz/alarm.go new file mode 100644 index 0000000..09945e2 --- /dev/null +++ b/internal/biz/alarm.go @@ -0,0 +1,69 @@ +package biz + +import ( + "context" + "fmt" + "github.com/redis/go-redis/v9" + "voucher/internal/biz/bo" + "voucher/internal/biz/vo" + "voucher/internal/pkg/lock" +) + +func (v *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() + + if err == nil { + // 缓存存在，直接返回 + return nil + } + + if err != redis.Nil { + return fmt.Errorf(fmt.Sprintf("alarm 获取redis缓存%s异常:%v", 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 { + // 二次获取，判定处理，以免获取锁后又执行了一次 + + cacheValue, err3 := v.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)) + } + + if len(cacheValue) > 0 { + return nil // 有直接返回 + } + + // 通知 + if err = v.DingMixRepo.SendMarkdownMessage(ctx, "异常通知", v.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)) + } + + return nil + }) +} + +func (v *VoucherBiz) alarmText(_ context.Context, order *bo.OrderBo, errMsg string) string { + + remarks := fmt.Sprintf("订单号:%s，商品编号:%s,原因:%s", order.OrderNo, order.ProductNo, errMsg) + + msg := "# " + + "

立减金发放平台报警通知

" + + " \n" + + "" + + "不好了，订单发放发生异常了" + + "[%s]请尽快处理@相关人员。" + + "" + + return fmt.Sprintf(msg, remarks) +} diff --git a/internal/biz/order.go b/internal/biz/order.go index b1bd4ed..29db30e 100644 --- a/internal/biz/order.go +++ b/internal/biz/order.go @@ -3,12 +3,9 @@ package biz import ( "context" "fmt" - "github.com/go-kratos/kratos/v2/log" - "github.com/redis/go-redis/v9" err2 "voucher/api/err" "voucher/internal/biz/bo" "voucher/internal/biz/vo" - "voucher/internal/pkg/lock" ) func (v *VoucherBiz) OrderRetry(ctx context.Context, outBizNos []string) error { @@ -106,102 +103,13 @@ func (v *VoucherBiz) create(ctx context.Context, req *bo.OrderCreateReqBo, produ return v.OrderRepo.Create(ctx, o) } -// RegisterTag 注册通知标签 stock.MchId 批次创建商户, stock.BatchNo 商品批次号 -func (this *VoucherBiz) RegisterTag(ctx context.Context, productNo string) error { - - stock, err := this.ProductRepo.GetByProductNo(ctx, productNo) - if err != nil { - return err - } - - return this.registerNotifyTag(ctx, stock.MchId, stock.BatchNo) -} - -func (v *VoucherBiz) registerNotifyTag(ctx context.Context, stockCreatorMchID, stockID string) error { - - c := vo.WechatNotifyRegisterTagCacheKey.BuildCache([]string{v.bc.WechatNotifyMQ.Tag, stockCreatorMchID, stockID}) - - _, err := v.rdb.Rdb.Get(ctx, c.Key).Result() - - if err == nil { - // 缓存存在，直接返回 - return nil - } - - if err != redis.Nil { - return fmt.Errorf(fmt.Sprintf("获取redis缓存%s异常:%v", c.Key, err)) - } - - cl := vo.WechatNotifyRegisterTagCacheLockKey.BuildCache([]string{v.bc.WechatNotifyMQ.Tag, stockCreatorMchID, stockID}) - - return lock.NewMutex(v.rdb.Rdb, cl.TTL).Lock(ctx, cl.Key, func(ctx context.Context) error { - // 二次获取，判定处理，以免获取锁后又执行了一次 - - cacheValue, err3 := v.rdb.Rdb.Get(ctx, c.Key).Result() - - if err3 != nil && err3 != redis.Nil { - return fmt.Errorf(fmt.Sprintf("二次获取redis缓存%s异常:%v", c.Key, err)) - } - - if cacheValue != "" { - return nil // 有直接返回 - } - - wechatNotifyTag, err3 := v.WechatNotifyRegisterTagRepo.GetByStockIdAndMchId(ctx, stockCreatorMchID, stockID) - if err3 != nil && !err2.IsDbNotFound(err3) { - return err3 - } - - if wechatNotifyTag != nil { - if wechatNotifyTag.Tag != v.bc.WechatNotifyMQ.Tag { - return fmt.Errorf("tag不一致，请检查tag配置:%s", wechatNotifyTag.Tag) - } - - if wechatNotifyTag.Status.IsSuccess() { - return v.setCache(ctx, c, wechatNotifyTag) - } - } else { - wechatNotifyTag, err3 = v.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 = v.WechatNotifyRegisterTagRepo.Success(ctx, wechatNotifyTag.ID); err != nil { - return err - } - - return v.setCache(ctx, c, wechatNotifyTag) - }) -} - -func (v *VoucherBiz) createWechatNotifyRegisterTag(ctx context.Context, stockCreatorMchID, stockID string) (*bo.WechatNotifyRegisterTagBo, error) { - return v.WechatNotifyRegisterTagRepo.Create(ctx, &bo.WechatNotifyRegisterTagBo{ - StockID: stockID, - StockCreatorMchID: stockCreatorMchID, - Tag: v.bc.WechatNotifyMQ.Tag, - }) -} - -func (v *VoucherBiz) setCache(ctx context.Context, c *vo.Cache, wechatNotifyTag *bo.WechatNotifyRegisterTagBo) error { - - if err := v.rdb.Rdb.Set(ctx, c.Key, wechatNotifyTag.Tag, c.TTL).Err(); err != nil { - return fmt.Errorf(fmt.Sprintf("设置redis缓存%s异常:%v", c.Key, err)) - } - - return nil -} - func (v *VoucherBiz) ing(ctx context.Context, id uint64) error { + return v.OrderRepo.Ing(ctx, id) } func (v *VoucherBiz) success(ctx context.Context, order *bo.OrderBo, voucherNo string) error { + return v.OrderRepo.Success(ctx, order.ID, voucherNo) } @@ -218,86 +126,6 @@ func (v *VoucherBiz) fail(ctx context.Context, order *bo.OrderBo, errReq error) return v.alarm(ctx, order, errReq.Error()) } -func (v *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() - - if err == nil { - // 缓存存在，直接返回 - return nil - } - - if err != redis.Nil { - return fmt.Errorf(fmt.Sprintf("alarm 获取redis缓存%s异常:%v", 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 { - // 二次获取，判定处理，以免获取锁后又执行了一次 - - cacheValue, err3 := v.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)) - } - - if len(cacheValue) > 0 { - return nil // 有直接返回 - } - - // 通知 - if err = v.DingMixRepo.SendMarkdownMessage(ctx, "异常通知", v.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)) - } - - return nil - }) -} - -func (v *VoucherBiz) alarmText(_ context.Context, order *bo.OrderBo, errMsg string) string { - - remarks := fmt.Sprintf("订单号:%s，商品编号:%s,原因:%s", order.OrderNo, order.ProductNo, errMsg) - - msg := "# " + - "

立减金发放平台报警通知

" + - " \n" + - "" + - "不好了，订单发放发生异常了" + - "[%s]请尽快处理@相关人员。" + - "" - - return fmt.Sprintf(msg, remarks) -} - -func (v *VoucherBiz) Query(ctx context.Context, order *bo.OrderBo) error { - - status, err := v.WechatCpnRepo.Query(ctx, order) - if err != nil { - return err - } - - if order.Status == status { - log.Warnf("券状态未改变:%s，忽略不处理,orderNo:%s", order.Status.GetText(), order.OrderNo) - return nil - } - - if err = v.UpdateOrderStatus(ctx, order.ID, status); err != nil { - return err - } - - order.Status = status - - return nil -} - func (v *VoucherBiz) UpdateOrderStatus(ctx context.Context, orderId uint64, status vo.OrderStatus) error { if status.IsSuccess() { @@ -315,18 +143,3 @@ func (v *VoucherBiz) UpdateOrderStatus(ctx context.Context, orderId uint64, stat return fmt.Errorf("notice 未知券状态,orderId:%d,statuText:%s", orderId, status.GetText()) } - -func (v *VoucherBiz) QueryOrder(ctx context.Context, orderNo string) (string, error) { - - order, err3 := v.OrderRepo.GetByOrderNo(ctx, orderNo) - if err3 != nil { - return "", err3 - } - - status, err := v.WechatCpnRepo.Query(ctx, order) - if err != nil { - return "", err - } - - return fmt.Sprintf("orderNo:%s,订单状态:%s,微信查询返回状态:%s", orderNo, order.Status.GetText(), status.GetText()), nil -} diff --git a/internal/biz/query.go b/internal/biz/query.go new file mode 100644 index 0000000..5bdc0af --- /dev/null +++ b/internal/biz/query.go @@ -0,0 +1,44 @@ +package biz + +import ( + "context" + "fmt" + "github.com/go-kratos/kratos/v2/log" + "voucher/internal/biz/bo" +) + +func (v *VoucherBiz) Query(ctx context.Context, order *bo.OrderBo) error { + + status, err := v.WechatCpnRepo.Query(ctx, order) + if err != nil { + return err + } + + if order.Status == status { + log.Warnf("券状态未改变:%s，忽略不处理,orderNo:%s", order.Status.GetText(), order.OrderNo) + return nil + } + + if err = v.UpdateOrderStatus(ctx, order.ID, status); err != nil { + return err + } + + order.Status = status + + return nil +} + +func (v *VoucherBiz) QueryOrder(ctx context.Context, orderNo string) (string, error) { + + order, err3 := v.OrderRepo.GetByOrderNo(ctx, orderNo) + if err3 != nil { + return "", err3 + } + + status, err := v.WechatCpnRepo.Query(ctx, order) + if err != nil { + return "", err + } + + return fmt.Sprintf("orderNo:%s,订单状态:%s,微信查询返回状态:%s", orderNo, order.Status.GetText(), status.GetText()), nil +} diff --git a/internal/biz/register_tag.go b/internal/biz/register_tag.go new file mode 100644 index 0000000..cb8314d --- /dev/null +++ b/internal/biz/register_tag.go @@ -0,0 +1,102 @@ +package biz + +import ( + "context" + "fmt" + "github.com/redis/go-redis/v9" + err2 "voucher/api/err" + "voucher/internal/biz/bo" + "voucher/internal/biz/vo" + "voucher/internal/pkg/lock" +) + +// RegisterTag 注册通知标签 stock.MchId 批次创建商户, stock.BatchNo 商品批次号 +func (this *VoucherBiz) RegisterTag(ctx context.Context, productNo string) error { + + stock, err := this.ProductRepo.GetByProductNo(ctx, productNo) + if err != nil { + return err + } + + return this.registerNotifyTag(ctx, stock.MchId, stock.BatchNo) +} + +func (v *VoucherBiz) registerNotifyTag(ctx context.Context, stockCreatorMchID, stockID string) error { + + c := vo.WechatNotifyRegisterTagCacheKey.BuildCache([]string{v.bc.WechatNotifyMQ.Tag, stockCreatorMchID, stockID}) + + _, err := v.rdb.Rdb.Get(ctx, c.Key).Result() + + if err == nil { + // 缓存存在，直接返回 + return nil + } + + if err != redis.Nil { + return fmt.Errorf(fmt.Sprintf("获取redis缓存%s异常:%v", c.Key, err)) + } + + cl := vo.WechatNotifyRegisterTagCacheLockKey.BuildCache([]string{v.bc.WechatNotifyMQ.Tag, stockCreatorMchID, stockID}) + + return lock.NewMutex(v.rdb.Rdb, cl.TTL).Lock(ctx, cl.Key, func(ctx context.Context) error { + // 二次获取，判定处理，以免获取锁后又执行了一次 + + cacheValue, err3 := v.rdb.Rdb.Get(ctx, c.Key).Result() + + if err3 != nil && err3 != redis.Nil { + return fmt.Errorf(fmt.Sprintf("二次获取redis缓存%s异常:%v", c.Key, err)) + } + + if cacheValue != "" { + return nil // 有直接返回 + } + + wechatNotifyTag, err3 := v.WechatNotifyRegisterTagRepo.GetByStockIdAndMchId(ctx, stockCreatorMchID, stockID) + if err3 != nil && !err2.IsDbNotFound(err3) { + return err3 + } + + if wechatNotifyTag != nil { + if wechatNotifyTag.Tag != v.bc.WechatNotifyMQ.Tag { + return fmt.Errorf("tag不一致，请检查tag配置:%s", wechatNotifyTag.Tag) + } + + if wechatNotifyTag.Status.IsSuccess() { + return v.setCache(ctx, c, wechatNotifyTag) + } + } else { + wechatNotifyTag, err3 = v.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 = v.WechatNotifyRegisterTagRepo.Success(ctx, wechatNotifyTag.ID); err != nil { + return err + } + + return v.setCache(ctx, c, wechatNotifyTag) + }) +} + +func (v *VoucherBiz) createWechatNotifyRegisterTag(ctx context.Context, stockCreatorMchID, stockID string) (*bo.WechatNotifyRegisterTagBo, error) { + return v.WechatNotifyRegisterTagRepo.Create(ctx, &bo.WechatNotifyRegisterTagBo{ + StockID: stockID, + StockCreatorMchID: stockCreatorMchID, + Tag: v.bc.WechatNotifyMQ.Tag, + }) +} + +func (v *VoucherBiz) setCache(ctx context.Context, c *vo.Cache, wechatNotifyTag *bo.WechatNotifyRegisterTagBo) error { + + if err := v.rdb.Rdb.Set(ctx, c.Key, wechatNotifyTag.Tag, c.TTL).Err(); err != nil { + return fmt.Errorf(fmt.Sprintf("设置redis缓存%s异常:%v", c.Key, err)) + } + + return nil +}