notice

configs/config.yaml

@@ -66,6 +66,8 @@ cmb:
   cmbKeyAlias: "SM2_CMBLIFE"
   orgNo: "LANSEXIONGDI" # 发码机构号，固定值，掌上生活优惠券系统提供
   notifyUrl: "https://sandbox.cdcc.cmbchina.com/AccessGateway/transIn/updateCodeStatus.json" # 招行测试回调地址
+  noticeStartDays: 7
+  noticeEndDays: 1
 
 #告警配置
 alarm:

internal/biz/bo/order_bo.go

@@ -38,6 +38,8 @@ type OrderCreateReqBo struct {
 	Attach      string
 }
 
-type OrderCreateRepBo struct {
+type FindInBatchesUseBo struct {
-	OrderNo string
+	Type      vo.OrderType
+	StartTime *time.Time
+	EndTime   *time.Time
 }

internal/biz/cmb/notify.go

@@ -46,9 +46,9 @@ func (v *Cmb) Notify(ctx context.Context, order *bo.OrderBo) (*bo.OrderNotifyBo,
 		return orderNotify, v.notifyFail(ctx, orderNotify.ID, err.Error())
 	}
 
-	replyJson, _ := json.Marshal(reply)
-
 	if reply.RespCode == vo.CmbResponseStatusSuccess.GetValue() {
+
+		replyJson, _ := json.Marshal(reply)
 		return orderNotify, v.notifySuccess(ctx, orderNotify.ID, string(replyJson))
 	}
 
@@ -79,17 +79,27 @@ func (v *Cmb) bizContent(_ context.Context, order *bo.OrderBo, orderNotify *bo.O
 	return string(bizJsonBytes), nil
 }
 
-func (v *Cmb) notifyCreate(ctx context.Context, order *bo.OrderBo, req *bo.OrderNotifyBo) (*v1.CmbRequest, *bo.OrderNotifyBo, error) {
+func (v *Cmb) NotifyRequest(ctx context.Context, order *bo.OrderBo, req *bo.OrderNotifyBo) (*v1.CmbRequest, error) {
 
 	bizContent, err := v.bizContent(ctx, order, req)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
 	request, err := v.CmbMixRepo.GetRequest(ctx, &bo.CmbRequestBo{
 		FuncName:   vo.CmbNotifyFuncName,
 		BizContent: bizContent,
 	})
+	if err != nil {
+		return nil, err
+	}
+
+	return request, nil
+}
+
+func (v *Cmb) notifyCreate(ctx context.Context, order *bo.OrderBo, req *bo.OrderNotifyBo) (*v1.CmbRequest, *bo.OrderNotifyBo, error) {
+
+	request, err := v.NotifyRequest(ctx, order, req)
 	if err != nil {
 		return nil, nil, err
 	}

internal/biz/order_notice.go

@@ -0,0 +1,164 @@
+package biz
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/go-kratos/kratos/v2/log"
+	"github.com/redis/go-redis/v9"
+	"time"
+	"voucher/internal/biz/bo"
+	"voucher/internal/biz/vo"
+	"voucher/internal/pkg/lock"
+)
+
+func (v *VoucherBiz) OrderNotice(ctx context.Context) error {
+
+	if err := v.isCanNotice(ctx); err != nil {
+		return err
+	}
+
+	now := time.Now()
+
+	// 获取七天前的日期
+	sevenDaysAgo := now.AddDate(0, 0, int(-v.bc.Cmb.NoticeStartDays))
+	// 获取七天前 00:00:00 的时间
+	startTime := time.Date(sevenDaysAgo.Year(), sevenDaysAgo.Month(), sevenDaysAgo.Day(), 0, 0, 0, 0, sevenDaysAgo.Location())
+
+	sevenDaysAgoEnd := now.AddDate(0, 0, int(-v.bc.Cmb.NoticeEndDays))
+	// 获取昨天 23:59:59 的时间
+	endTime := time.Date(sevenDaysAgoEnd.Year(), sevenDaysAgoEnd.Month(), sevenDaysAgoEnd.Day(), 23, 59, 59, 0, sevenDaysAgoEnd.Location())
+
+	req := &bo.FindInBatchesUseBo{
+		Type:      vo.OrderTypeCmb,
+		StartTime: &startTime,
+		EndTime:   &endTime,
+	}
+
+	return v.OrderRepo.FindInBatches(ctx, req, func(ctx context.Context, rows []*bo.OrderBo) error {
+
+		for _, order := range rows {
+
+			if order.Type.IsCmb() {
+				if err := v.cmbOrderNotice(ctx, order); err != nil {
+					return err
+				}
+			}
+
+		}
+
+		return nil
+	})
+
+}
+
+func (v *VoucherBiz) isCanNotice(ctx context.Context) error {
+
+	if v.bc.Cmb.NoticeStartDays == 0 {
+		log.Warnf("noticeStartDays eq 0")
+		return nil
+	}
+
+	if v.bc.Cmb.NoticeEndDays == 0 {
+		log.Warnf("noticeEndDays eq 0")
+		return nil
+	}
+
+	cache := vo.CmbBatchNoticeCacheKey.BuildCache([]string{""})
+
+	_, err := v.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))
+	}
+
+	c := vo.CmbBatchNoticeLockKey.BuildCache([]string{""})
+
+	return lock.NewMutex(v.rdb.Rdb, c.TTL).Lock(ctx, c.Key, func(ctx context.Context) error {
+
+		// 二次获取，判定处理，以免获取锁后又执行了一次
+
+		cacheValue, err := v.rdb.Rdb.Get(ctx, cache.Key).Result()
+
+		if err != nil && err != redis.Nil {
+			return fmt.Errorf(fmt.Sprintf("notice 二次获取redis缓存%s异常:%v", cache.Key, err))
+		}
+
+		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))
+		}
+
+		log.Warnf("notice 获取redis缓存,不存在，开始处理")
+		return nil
+	})
+}
+
+func (v *VoucherBiz) cmbOrderNotice(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("notice 券状态未改变:%s，忽略不处理,orderNo:%s", order.Status.GetText(), order.OrderNo)
+		return nil
+	}
+
+	if status.IsSuccess() {
+		if err = v.OrderRepo.Available(ctx, order.ID); err != nil {
+			return err
+		}
+	} else if status.IsUse() {
+		if err = v.OrderRepo.Used(ctx, order.ID); err != nil {
+			return err
+		}
+	} else if status.IsExpired() {
+		if err = v.OrderRepo.Expired(ctx, order.ID); err != nil {
+			return err
+		}
+	} else {
+		return fmt.Errorf("notice 未知券状态,orderNo:%s,statuText:%s", order.OrderNo, status.GetText())
+	}
+
+	order.Status = status
+
+	event, err := order.Status.GetOrderNotifyEvent()
+	if err != nil {
+		return err
+	}
+
+	req := &bo.OrderNotifyBo{
+		OrderNo:   order.OrderNo,
+		NotifyUrl: order.NotifyUrl,
+		Channel:   order.Channel,
+		Event:     event,
+		Type:      order.Type,
+	}
+
+	request, err := v.Cmb.NotifyRequest(ctx, order, req)
+	if err != nil {
+		return err
+	}
+
+	reply, err := v.CmbMixRepo.Request(ctx, request, order.NotifyUrl)
+	if err != nil {
+		return err
+	}
+
+	if reply.RespCode != vo.CmbResponseStatusSuccess.GetValue() {
+		replyJson, _ := json.Marshal(reply)
+		return fmt.Errorf("callback notify cmb error,orderNo:%s,resp:%s", order.OrderNo, string(replyJson))
+	}
+
+	return nil
+}

internal/biz/repo/order.go

@@ -7,6 +7,7 @@ import (
 )
 
 type OrderRepo interface {
+	FindInBatches(ctx context.Context, w *bo.FindInBatchesUseBo, fun func(ctx context.Context, rows []*bo.OrderBo) error) error
 	GetByOutBizNo(ctx context.Context, t vo.OrderType, outBizNo string) (*bo.OrderBo, error)
 	GetByOrderNo(ctx context.Context, orderNo string) (*bo.OrderBo, error)
 	GetByMBV(ctx context.Context, merchantNo, batchNo, voucherNo string) (*bo.OrderBo, error)

internal/biz/vo/cache.go

@@ -11,6 +11,8 @@ const (
 	CmbOrderLockKey        CacheKey = "cmb_order"
 	CmbQueryLockKey        CacheKey = "cmb_query"
 	CmbProductQueryLockKey CacheKey = "cmb_product_query"
+	CmbBatchNoticeCacheKey CacheKey = "cmb_batch_notice"
+	CmbBatchNoticeLockKey  CacheKey = "cmb_batch_notice_lock"
 
 	NotifyRetryConsume CacheKey = "notify_retry_consume"
 
@@ -27,6 +29,8 @@ var CacheKeyMap = map[CacheKey]time.Duration{
 	CmbOrderLockKey:                     30 * time.Second,
 	CmbQueryLockKey:                     30 * time.Second,
 	CmbProductQueryLockKey:              30 * time.Second,
+	CmbBatchNoticeCacheKey:              43200 * time.Second,
+	CmbBatchNoticeLockKey:               300 * time.Second,
 	OrderConsumeFailAlarmKey:            3600 * time.Second, // 1小时
 	OrderConsumeFailAlarmLockKey:        60 * time.Second,
 	NotifyRetryConsume:                  60 * time.Second,

internal/biz/voucher.go

@@ -21,6 +21,7 @@ type VoucherBiz struct {
 	GenerateMixRepo             mixrepos.GenerateMixRepo
 	WechatCpnRepo               wechatrepo.WechatCpnRepo
 	DingMixRepo                 mixrepos.DingMixRepo
+	CmbMixRepo                  mixrepos.CmbMixRepo
 }
 
 func NewVoucherBiz(
@@ -35,6 +36,7 @@ func NewVoucherBiz(
 	GenerateMixRepo mixrepos.GenerateMixRepo,
 	WechatCpnRepo wechatrepo.WechatCpnRepo,
 	DingMixRepo mixrepos.DingMixRepo,
+	CmbMixRepo mixrepos.CmbMixRepo,
 ) *VoucherBiz {
 	return &VoucherBiz{
 		bc:                          bc,
@@ -48,5 +50,6 @@ func NewVoucherBiz(
 		GenerateMixRepo:             GenerateMixRepo,
 		WechatCpnRepo:               WechatCpnRepo,
 		DingMixRepo:                 DingMixRepo,
+		CmbMixRepo:                  CmbMixRepo,
 	}
 }

internal/conf/conf.pb.go

@@ -454,6 +454,8 @@ type Cmb struct {
 	CmbKeyAlias     string `protobuf:"bytes,8,opt,name=cmbKeyAlias,proto3" json:"cmbKeyAlias,omitempty"`
 	OrgNo           string `protobuf:"bytes,9,opt,name=orgNo,proto3" json:"orgNo,omitempty"`
 	NotifyUrl       string `protobuf:"bytes,10,opt,name=notifyUrl,proto3" json:"notifyUrl,omitempty"`
+	NoticeStartDays int64  `protobuf:"varint,11,opt,name=noticeStartDays,proto3" json:"noticeStartDays,omitempty"`
+	NoticeEndDays   int64  `protobuf:"varint,12,opt,name=noticeEndDays,proto3" json:"noticeEndDays,omitempty"`
 }
 
 func (x *Cmb) Reset() {
@@ -558,6 +560,20 @@ func (x *Cmb) GetNotifyUrl() string {
 	return ""
 }
 
+func (x *Cmb) GetNoticeStartDays() int64 {
+	if x != nil {
+		return x.NoticeStartDays
+	}
+	return 0
+}
+
+func (x *Cmb) GetNoticeEndDays() int64 {
+	if x != nil {
+		return x.NoticeEndDays
+	}
+	return 0
+}
+
 type WechatNotifyMQ struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -1322,7 +1338,7 @@ var file_conf_conf_proto_rawDesc = []byte{
 	0x1a, 0x6d, 0x63, 0x68, 0x43, 0x65, 0x72, 0x74, 0x69, 0x66, 0x69, 0x63, 0x61, 0x74, 0x65, 0x53,
 	0x65, 0x72, 0x69, 0x61, 0x6c, 0x4e, 0x75, 0x6d, 0x62, 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28,
 	0x09, 0x52, 0x1a, 0x6d, 0x63, 0x68, 0x43, 0x65, 0x72, 0x74, 0x69, 0x66, 0x69, 0x63, 0x61, 0x74,
-	0x65, 0x53, 0x65, 0x72, 0x69, 0x61, 0x6c, 0x4e, 0x75, 0x6d, 0x62, 0x65, 0x72, 0x22, 0x87, 0x02,
+	0x65, 0x53, 0x65, 0x72, 0x69, 0x61, 0x6c, 0x4e, 0x75, 0x6d, 0x62, 0x65, 0x72, 0x22, 0xd7, 0x02,
 	0x0a, 0x03, 0x43, 0x6d, 0x62, 0x12, 0x10, 0x0a, 0x03, 0x6d, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01,
 	0x28, 0x09, 0x52, 0x03, 0x6d, 0x69, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x61, 0x69, 0x64, 0x18, 0x02,
 	0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x61, 0x69, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x6d, 0x32,
@@ -1339,7 +1355,12 @@ var file_conf_conf_proto_rawDesc = []byte{
 	0x69, 0x61, 0x73, 0x12, 0x14, 0x0a, 0x05, 0x6f, 0x72, 0x67, 0x4e, 0x6f, 0x18, 0x09, 0x20, 0x01,
 	0x28, 0x09, 0x52, 0x05, 0x6f, 0x72, 0x67, 0x4e, 0x6f, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x6f, 0x74,
 	0x69, 0x66, 0x79, 0x55, 0x72, 0x6c, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x6f,
-	0x74, 0x69, 0x66, 0x79, 0x55, 0x72, 0x6c, 0x22, 0xda, 0x02, 0x0a, 0x0e, 0x57, 0x65, 0x63, 0x68,
+	0x74, 0x69, 0x66, 0x79, 0x55, 0x72, 0x6c, 0x12, 0x28, 0x0a, 0x0f, 0x6e, 0x6f, 0x74, 0x69, 0x63,
+	0x65, 0x53, 0x74, 0x61, 0x72, 0x74, 0x44, 0x61, 0x79, 0x73, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x03,
+	0x52, 0x0f, 0x6e, 0x6f, 0x74, 0x69, 0x63, 0x65, 0x53, 0x74, 0x61, 0x72, 0x74, 0x44, 0x61, 0x79,
+	0x73, 0x12, 0x24, 0x0a, 0x0d, 0x6e, 0x6f, 0x74, 0x69, 0x63, 0x65, 0x45, 0x6e, 0x64, 0x44, 0x61,
+	0x79, 0x73, 0x18, 0x0c, 0x20, 0x01, 0x28, 0x03, 0x52, 0x0d, 0x6e, 0x6f, 0x74, 0x69, 0x63, 0x65,
+	0x45, 0x6e, 0x64, 0x44, 0x61, 0x79, 0x73, 0x22, 0xda, 0x02, 0x0a, 0x0e, 0x57, 0x65, 0x63, 0x68,
 	0x61, 0x74, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x4d, 0x51, 0x12, 0x20, 0x0a, 0x0b, 0x61, 0x63,
 	0x63, 0x65, 0x73, 0x73, 0x4b, 0x65, 0x79, 0x49, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
 	0x0b, 0x61, 0x63, 0x63, 0x65, 0x73, 0x73, 0x4b, 0x65, 0x79, 0x49, 0x64, 0x12, 0x28, 0x0a, 0x0f,

internal/conf/conf.proto

@@ -84,6 +84,8 @@ message Cmb {
   string cmbKeyAlias = 8;
   string orgNo = 9;
   string notifyUrl = 10;
+  int64 noticeStartDays = 11;
+  int64 noticeEndDays = 12;
 }
 
 message WechatNotifyMQ {

internal/data/repoimpl/order.go

@@ -27,6 +27,32 @@ func (p *OrderRepoImpl) DB(ctx context.Context) *gorm.DB {
 	return p.db.DB(ctx).Model(model.Order{})
 }
 
+func (p *OrderRepoImpl) FindInBatches(ctx context.Context, w *bo.FindInBatchesUseBo, fun func(ctx context.Context, rows []*bo.OrderBo) error) error {
+
+	var results = make([]*model.Order, 0)
+
+	db := p.db.DB(ctx)
+
+	db.Where("status = ?", vo.OrderStatusUse)
+	db.Where("type = ?", w.Type)
+	db.Where("last_use_time >= ?", w.StartTime)
+	db.Where("last_use_time <= ?", w.EndTime)
+
+	// 处理记录，批处理大小为100
+	result := db.FindInBatches(&results, 100, func(tx *gorm.DB, batch int) error {
+		// tx.RowsAffected 提供当前批处理中记录的计数（the count of records in the current batch）
+		// 'batch' 变量表示当前批号（the current batch number）
+		// 返回 error 将阻止更多的批处理
+		return fun(ctx, p.ToBos(results))
+	})
+
+	if result.Error != nil {
+		return result.Error
+	}
+
+	return nil
+}
+
 func (p *OrderRepoImpl) Create(ctx context.Context, req *bo.OrderBo) (*bo.OrderBo, error) {
 	now := time.Now()
 

internal/pkg/helper/int_test.go

@@ -3,6 +3,7 @@ package helper
 import (
 	"fmt"
 	"testing"
+	"time"
 )
 
 func TestBuildStr(t *testing.T) {
@@ -15,3 +16,33 @@ func TestBuildStr(t *testing.T) {
 	resultStr := BuildStr(uid, arrStr)
 	fmt.Println(resultStr)
 }
+
+func TestAddDate(t *testing.T) {
+	// 获取当前时间
+	now := time.Now()
+	fmt.Println("当前时间:", now.Format(time.DateTime))
+
+	// 当前时间往后推 7 天
+	future := now.AddDate(0, 0, 7)
+	fmt.Println("当前时间往后推 7 天:", future.Format(time.DateTime))
+
+	// 当前时间往前推 7 天
+	past := now.AddDate(0, 0, -7)
+	fmt.Println("当前时间往前推 7 天:", past.Format(time.DateTime))
+}
+
+func TestSevenDaysAgo(t *testing.T) {
+	// 获取当前时间
+	now := time.Now()
+
+	sevenDaysAgoEnd := now.AddDate(0, 0, -1)
+	// 获取今天 23:59:59 的时间
+	endTime := time.Date(sevenDaysAgoEnd.Year(), sevenDaysAgoEnd.Month(), sevenDaysAgoEnd.Day(), 23, 59, 59, 0, sevenDaysAgoEnd.Location())
+	// 获取七天前的日期
+	sevenDaysAgo := now.AddDate(0, 0, -7)
+	// 获取七天前 00:00:00 的时间
+	startTime := time.Date(sevenDaysAgo.Year(), sevenDaysAgo.Month(), sevenDaysAgo.Day(), 0, 0, 0, 0, sevenDaysAgo.Location())
+
+	fmt.Printf("开始时间: %s\n", startTime.Format(time.DateTime))
+	fmt.Printf("结束时间: %s\n", endTime.Format(time.DateTime))
+}

internal/server/cron.go

@@ -33,7 +33,7 @@ func (cs *CronServer) Start(ctx context.Context) error {
 		return nil
 	}
 
-	if err := cs.VoucherService.CmbOrderNotice(ctx); err != nil {
+	if err := cs.VoucherService.OrderNotice(ctx); err != nil {
 		log.Error("cron order notice Start err: %v", err)
 		return err
 	}

internal/service/cmb_order_notice.go

@@ -1,8 +0,0 @@
-package service
-
-import "context"
-
-func (s *VoucherService) CmbOrderNotice(ctx context.Context) error {
-
-	return nil
-}

internal/service/voucher.go

@@ -1,6 +1,7 @@
 package service
 
 import (
+	"context"
 	"voucher/internal/biz"
 	"voucher/internal/biz/mixrepos"
 	"voucher/internal/biz/wechatrepo"
@@ -27,3 +28,8 @@ func NewVoucherService(
 		WechatCpnRepo: WechatCpnRepo,
 	}
 }
+
+func (s *VoucherService) OrderNotice(ctx context.Context) error {
+
+	return s.VoucherBiz.OrderNotice(ctx)
+}