feat: 实现订单超时自动取消功能,支持钱包余额解冻和 Asynq Scheduler 统一调度
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 6m58s
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 6m58s
- 新增 expires_at 字段和复合索引,待支付订单 30 分钟超时自动取消 - 实现 cancelOrder/unfreezeWalletForCancel 钱包余额解冻逻辑 - 创建 Asynq 定时任务(order_expire/alert_check/data_cleanup) - 将原有 time.Ticker 轮询迁移至 Asynq Scheduler 统一调度 - 同步 delta specs 到 main specs 并归档变更
This commit is contained in:
@@ -3,6 +3,7 @@ package bootstrap
|
||||
import (
|
||||
"github.com/break/junhong_cmp_fiber/internal/service/commission_calculation"
|
||||
"github.com/break/junhong_cmp_fiber/internal/service/commission_stats"
|
||||
orderSvc "github.com/break/junhong_cmp_fiber/internal/service/order"
|
||||
packagepkg "github.com/break/junhong_cmp_fiber/internal/service/package"
|
||||
pollingSvc "github.com/break/junhong_cmp_fiber/internal/service/polling"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/queue"
|
||||
@@ -77,6 +78,27 @@ func initWorkerServices(stores *queue.WorkerStores, deps *WorkerDependencies) *q
|
||||
deps.Logger,
|
||||
)
|
||||
|
||||
// 初始化订单服务(仅用于超时自动取消,不需要微信支付和队列客户端)
|
||||
orderService := orderSvc.New(
|
||||
deps.DB,
|
||||
deps.Redis,
|
||||
stores.Order,
|
||||
stores.OrderItem,
|
||||
stores.AgentWallet,
|
||||
stores.CardWallet,
|
||||
nil, // purchaseValidationService: 超时取消不需要
|
||||
stores.ShopPackageAllocation,
|
||||
stores.ShopSeriesAllocation,
|
||||
stores.IotCard,
|
||||
stores.Device,
|
||||
stores.PackageSeries,
|
||||
stores.PackageUsage,
|
||||
stores.Package,
|
||||
nil, // wechatPayment: 超时取消不需要
|
||||
nil, // queueClient: 超时取消不触发分佣
|
||||
deps.Logger,
|
||||
)
|
||||
|
||||
return &queue.WorkerServices{
|
||||
CommissionCalculation: commissionCalculationService,
|
||||
CommissionStats: commissionStatsService,
|
||||
@@ -85,5 +107,6 @@ func initWorkerServices(stores *queue.WorkerStores, deps *WorkerDependencies) *q
|
||||
ResetService: resetService,
|
||||
AlertService: alertService,
|
||||
CleanupService: cleanupService,
|
||||
OrderExpirer: orderService,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -28,6 +28,7 @@ type workerStores struct {
|
||||
DataCleanupLog *postgres.DataCleanupLogStore
|
||||
AgentWallet *postgres.AgentWalletStore
|
||||
AgentWalletTransaction *postgres.AgentWalletTransactionStore
|
||||
CardWallet *postgres.CardWalletStore
|
||||
}
|
||||
|
||||
func initWorkerStores(deps *WorkerDependencies) *queue.WorkerStores {
|
||||
@@ -54,6 +55,7 @@ func initWorkerStores(deps *WorkerDependencies) *queue.WorkerStores {
|
||||
DataCleanupLog: postgres.NewDataCleanupLogStore(deps.DB),
|
||||
AgentWallet: postgres.NewAgentWalletStore(deps.DB, deps.Redis),
|
||||
AgentWalletTransaction: postgres.NewAgentWalletTransactionStore(deps.DB, deps.Redis),
|
||||
CardWallet: postgres.NewCardWalletStore(deps.DB, deps.Redis),
|
||||
}
|
||||
|
||||
return &queue.WorkerStores{
|
||||
@@ -79,5 +81,6 @@ func initWorkerStores(deps *WorkerDependencies) *queue.WorkerStores {
|
||||
DataCleanupLog: stores.DataCleanupLog,
|
||||
AgentWallet: stores.AgentWallet,
|
||||
AgentWalletTransaction: stores.AgentWalletTransaction,
|
||||
CardWallet: stores.CardWallet,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -37,27 +37,27 @@ func (w *AgentWallet) GetAvailableBalance() int64 {
|
||||
// AgentWalletTransaction 代理钱包交易记录模型
|
||||
// 记录所有代理钱包余额变动
|
||||
type AgentWalletTransaction struct {
|
||||
ID uint `gorm:"column:id;primaryKey" json:"id"`
|
||||
AgentWalletID uint `gorm:"column:agent_wallet_id;not null;index;comment:代理钱包ID" json:"agent_wallet_id"`
|
||||
ShopID uint `gorm:"column:shop_id;not null;index;comment:店铺ID(冗余字段,便于查询)" json:"shop_id"`
|
||||
UserID uint `gorm:"column:user_id;not null;comment:操作人用户ID" json:"user_id"`
|
||||
TransactionType string `gorm:"column:transaction_type;type:varchar(20);not null;comment:交易类型(recharge-充值 | deduct-扣款 | refund-退款 | commission-分佣 | withdrawal-提现)" json:"transaction_type"`
|
||||
TransactionSubtype *string `gorm:"column:transaction_subtype;type:varchar(50);comment:交易子类型(细分 order_payment 场景)" json:"transaction_subtype,omitempty"`
|
||||
Amount int64 `gorm:"column:amount;type:bigint;not null;comment:变动金额(单位:分,正数为增加,负数为减少)" json:"amount"`
|
||||
BalanceBefore int64 `gorm:"column:balance_before;type:bigint;not null;comment:变动前余额(单位:分)" json:"balance_before"`
|
||||
BalanceAfter int64 `gorm:"column:balance_after;type:bigint;not null;comment:变动后余额(单位:分)" json:"balance_after"`
|
||||
Status int `gorm:"column:status;type:int;not null;default:1;comment:交易状态(1-成功 2-失败 3-处理中)" json:"status"`
|
||||
ReferenceType *string `gorm:"column:reference_type;type:varchar(50);comment:关联业务类型(order | commission | withdrawal | topup)" json:"reference_type,omitempty"`
|
||||
ReferenceID *uint `gorm:"column:reference_id;comment:关联业务ID" json:"reference_id,omitempty"`
|
||||
RelatedShopID *uint `gorm:"column:related_shop_id;comment:关联店铺ID(代购时记录下级店铺)" json:"related_shop_id,omitempty"`
|
||||
Remark *string `gorm:"column:remark;type:text;comment:备注" json:"remark,omitempty"`
|
||||
Metadata *string `gorm:"column:metadata;type:jsonb;comment:扩展信息(如手续费、支付方式等)" json:"metadata,omitempty"`
|
||||
Creator uint `gorm:"column:creator;not null;comment:创建人ID" json:"creator"`
|
||||
ShopIDTag uint `gorm:"column:shop_id_tag;not null;index;comment:店铺ID标签(多租户过滤)" json:"shop_id_tag"`
|
||||
EnterpriseIDTag *uint `gorm:"column:enterprise_id_tag;index;comment:企业ID标签(多租户过滤)" json:"enterprise_id_tag,omitempty"`
|
||||
CreatedAt time.Time `gorm:"column:created_at;not null;default:CURRENT_TIMESTAMP" json:"created_at"`
|
||||
UpdatedAt time.Time `gorm:"column:updated_at;not null;default:CURRENT_TIMESTAMP" json:"updated_at"`
|
||||
DeletedAt gorm.DeletedAt `gorm:"column:deleted_at;index" json:"deleted_at,omitempty"`
|
||||
ID uint `gorm:"column:id;primaryKey" json:"id"`
|
||||
AgentWalletID uint `gorm:"column:agent_wallet_id;not null;index;comment:代理钱包ID" json:"agent_wallet_id"`
|
||||
ShopID uint `gorm:"column:shop_id;not null;index;comment:店铺ID(冗余字段,便于查询)" json:"shop_id"`
|
||||
UserID uint `gorm:"column:user_id;not null;comment:操作人用户ID" json:"user_id"`
|
||||
TransactionType string `gorm:"column:transaction_type;type:varchar(20);not null;comment:交易类型(recharge-充值 | deduct-扣款 | refund-退款 | commission-分佣 | withdrawal-提现)" json:"transaction_type"`
|
||||
TransactionSubtype *string `gorm:"column:transaction_subtype;type:varchar(50);comment:交易子类型(细分 order_payment 场景)" json:"transaction_subtype,omitempty"`
|
||||
Amount int64 `gorm:"column:amount;type:bigint;not null;comment:变动金额(单位:分,正数为增加,负数为减少)" json:"amount"`
|
||||
BalanceBefore int64 `gorm:"column:balance_before;type:bigint;not null;comment:变动前余额(单位:分)" json:"balance_before"`
|
||||
BalanceAfter int64 `gorm:"column:balance_after;type:bigint;not null;comment:变动后余额(单位:分)" json:"balance_after"`
|
||||
Status int `gorm:"column:status;type:int;not null;default:1;comment:交易状态(1-成功 2-失败 3-处理中)" json:"status"`
|
||||
ReferenceType *string `gorm:"column:reference_type;type:varchar(50);comment:关联业务类型(order | commission | withdrawal | topup)" json:"reference_type,omitempty"`
|
||||
ReferenceID *uint `gorm:"column:reference_id;comment:关联业务ID" json:"reference_id,omitempty"`
|
||||
RelatedShopID *uint `gorm:"column:related_shop_id;comment:关联店铺ID(代购时记录下级店铺)" json:"related_shop_id,omitempty"`
|
||||
Remark *string `gorm:"column:remark;type:text;comment:备注" json:"remark,omitempty"`
|
||||
Metadata *string `gorm:"column:metadata;type:jsonb;comment:扩展信息(如手续费、支付方式等)" json:"metadata,omitempty"`
|
||||
Creator uint `gorm:"column:creator;not null;comment:创建人ID" json:"creator"`
|
||||
ShopIDTag uint `gorm:"column:shop_id_tag;not null;index;comment:店铺ID标签(多租户过滤)" json:"shop_id_tag"`
|
||||
EnterpriseIDTag *uint `gorm:"column:enterprise_id_tag;index;comment:企业ID标签(多租户过滤)" json:"enterprise_id_tag,omitempty"`
|
||||
CreatedAt time.Time `gorm:"column:created_at;not null;default:CURRENT_TIMESTAMP" json:"created_at"`
|
||||
UpdatedAt time.Time `gorm:"column:updated_at;not null;default:CURRENT_TIMESTAMP" json:"updated_at"`
|
||||
DeletedAt gorm.DeletedAt `gorm:"column:deleted_at;index" json:"deleted_at,omitempty"`
|
||||
}
|
||||
|
||||
// TableName 指定表名
|
||||
|
||||
@@ -29,8 +29,8 @@ type StandaloneIotCardResponse struct {
|
||||
IMSI string `json:"imsi,omitempty" description:"IMSI"`
|
||||
MSISDN string `json:"msisdn,omitempty" description:"卡接入号"`
|
||||
BatchNo string `json:"batch_no,omitempty" description:"批次号"`
|
||||
Supplier string `json:"supplier,omitempty" description:"供应商"`
|
||||
Status int `json:"status" description:"状态 (1:在库, 2:已分销, 3:已激活, 4:已停用)"`
|
||||
Supplier string `json:"supplier,omitempty" description:"供应商"`
|
||||
Status int `json:"status" description:"状态 (1:在库, 2:已分销, 3:已激活, 4:已停用)"`
|
||||
ShopID *uint `json:"shop_id,omitempty" description:"店铺ID"`
|
||||
ShopName string `json:"shop_name,omitempty" description:"店铺名称"`
|
||||
ActivatedAt *time.Time `json:"activated_at,omitempty" description:"激活时间"`
|
||||
|
||||
@@ -28,6 +28,7 @@ type OrderListRequest struct {
|
||||
PurchaseRole string `json:"purchase_role" query:"purchase_role" validate:"omitempty,oneof=self_purchase purchased_by_parent purchased_by_platform purchase_for_subordinate" description:"订单角色 (self_purchase:自己购买, purchased_by_parent:上级代理购买, purchased_by_platform:平台代购, purchase_for_subordinate:给下级购买)"`
|
||||
StartTime *time.Time `json:"start_time" query:"start_time" description:"创建时间起始"`
|
||||
EndTime *time.Time `json:"end_time" query:"end_time" description:"创建时间结束"`
|
||||
IsExpired *bool `json:"is_expired" query:"is_expired" description:"是否已过期 (true:已过期, false:未过期)"`
|
||||
}
|
||||
|
||||
type PayOrderRequest struct {
|
||||
@@ -44,21 +45,21 @@ type OrderItemResponse struct {
|
||||
}
|
||||
|
||||
type OrderResponse struct {
|
||||
ID uint `json:"id" description:"订单ID"`
|
||||
OrderNo string `json:"order_no" description:"订单号"`
|
||||
OrderType string `json:"order_type" description:"订单类型 (single_card:单卡购买, device:设备购买)"`
|
||||
BuyerType string `json:"buyer_type" description:"买家类型 (personal:个人客户, agent:代理商)"`
|
||||
BuyerID uint `json:"buyer_id" description:"买家ID"`
|
||||
IotCardID *uint `json:"iot_card_id,omitempty" description:"IoT卡ID"`
|
||||
DeviceID *uint `json:"device_id,omitempty" description:"设备ID"`
|
||||
TotalAmount int64 `json:"total_amount" description:"订单总金额(分)"`
|
||||
PaymentMethod string `json:"payment_method,omitempty" description:"支付方式 (wallet:钱包支付, wechat:微信支付, alipay:支付宝支付)"`
|
||||
PaymentStatus int `json:"payment_status" description:"支付状态 (1:待支付, 2:已支付, 3:已取消, 4:已退款)"`
|
||||
PaymentStatusText string `json:"payment_status_text" description:"支付状态文本"`
|
||||
PaidAt *time.Time `json:"paid_at,omitempty" description:"支付时间"`
|
||||
IsPurchaseOnBehalf bool `json:"is_purchase_on_behalf" description:"是否为代购订单"`
|
||||
CommissionStatus int `json:"commission_status" description:"佣金状态 (1:待计算, 2:已计算)"`
|
||||
CommissionConfigVersion int `json:"commission_config_version" description:"佣金配置版本"`
|
||||
ID uint `json:"id" description:"订单ID"`
|
||||
OrderNo string `json:"order_no" description:"订单号"`
|
||||
OrderType string `json:"order_type" description:"订单类型 (single_card:单卡购买, device:设备购买)"`
|
||||
BuyerType string `json:"buyer_type" description:"买家类型 (personal:个人客户, agent:代理商)"`
|
||||
BuyerID uint `json:"buyer_id" description:"买家ID"`
|
||||
IotCardID *uint `json:"iot_card_id,omitempty" description:"IoT卡ID"`
|
||||
DeviceID *uint `json:"device_id,omitempty" description:"设备ID"`
|
||||
TotalAmount int64 `json:"total_amount" description:"订单总金额(分)"`
|
||||
PaymentMethod string `json:"payment_method,omitempty" description:"支付方式 (wallet:钱包支付, wechat:微信支付, alipay:支付宝支付)"`
|
||||
PaymentStatus int `json:"payment_status" description:"支付状态 (1:待支付, 2:已支付, 3:已取消, 4:已退款)"`
|
||||
PaymentStatusText string `json:"payment_status_text" description:"支付状态文本"`
|
||||
PaidAt *time.Time `json:"paid_at,omitempty" description:"支付时间"`
|
||||
IsPurchaseOnBehalf bool `json:"is_purchase_on_behalf" description:"是否为代购订单"`
|
||||
CommissionStatus int `json:"commission_status" description:"佣金状态 (1:待计算, 2:已计算)"`
|
||||
CommissionConfigVersion int `json:"commission_config_version" description:"佣金配置版本"`
|
||||
|
||||
// 操作者信息
|
||||
OperatorID *uint `json:"operator_id,omitempty" description:"操作者ID"`
|
||||
@@ -74,6 +75,10 @@ type OrderResponse struct {
|
||||
Items []*OrderItemResponse `json:"items" description:"订单明细列表"`
|
||||
CreatedAt time.Time `json:"created_at" description:"创建时间"`
|
||||
UpdatedAt time.Time `json:"updated_at" description:"更新时间"`
|
||||
|
||||
// 订单超时信息
|
||||
ExpiresAt *time.Time `json:"expires_at,omitempty" description:"订单过期时间"`
|
||||
IsExpired bool `json:"is_expired" description:"是否已过期"`
|
||||
}
|
||||
|
||||
type OrderListResponse struct {
|
||||
|
||||
@@ -22,8 +22,8 @@ type IotCard struct {
|
||||
IMSI string `gorm:"column:imsi;type:varchar(50);comment:IMSI" json:"imsi"`
|
||||
MSISDN string `gorm:"column:msisdn;type:varchar(20);comment:MSISDN(手机号码)" json:"msisdn"`
|
||||
BatchNo string `gorm:"column:batch_no;type:varchar(100);comment:批次号" json:"batch_no"`
|
||||
Supplier string `gorm:"column:supplier;type:varchar(255);comment:供应商" json:"supplier"`
|
||||
Status int `gorm:"column:status;type:int;default:1;not null;comment:状态 1-在库 2-已分销 3-已激活 4-已停用" json:"status"`
|
||||
Supplier string `gorm:"column:supplier;type:varchar(255);comment:供应商" json:"supplier"`
|
||||
Status int `gorm:"column:status;type:int;default:1;not null;comment:状态 1-在库 2-已分销 3-已激活 4-已停用" json:"status"`
|
||||
ShopID *uint `gorm:"column:shop_id;index;comment:店铺ID(NULL=平台所有,有值=店铺所有)" json:"shop_id,omitempty"`
|
||||
ActivatedAt *time.Time `gorm:"column:activated_at;comment:激活时间" json:"activated_at"`
|
||||
ActivationStatus int `gorm:"column:activation_status;type:int;default:0;not null;comment:激活状态 0-未激活 1-已激活" json:"activation_status"`
|
||||
|
||||
@@ -52,6 +52,9 @@ type Order struct {
|
||||
|
||||
// 订单角色(标识订单中的买卖关系)
|
||||
PurchaseRole string `gorm:"column:purchase_role;type:varchar(50);index:idx_orders_purchase_role;comment:订单角色(self_purchase/purchased_by_parent/purchased_by_platform/purchase_for_subordinate)" json:"purchase_role,omitempty"`
|
||||
|
||||
// 订单超时信息
|
||||
ExpiresAt *time.Time `gorm:"column:expires_at;comment:订单过期时间(NULL表示不过期)" json:"expires_at,omitempty"`
|
||||
}
|
||||
|
||||
// TableName 指定表名
|
||||
|
||||
@@ -477,7 +477,6 @@ func (s *Service) GetDeviceDetail(ctx context.Context, deviceID uint) (*dto.Ente
|
||||
return nil, errors.New(errors.CodeDeviceNotAuthorized, "设备未授权给此企业")
|
||||
}
|
||||
|
||||
|
||||
var device model.Device
|
||||
if err := s.db.WithContext(ctx).Where("id = ?", deviceID).First(&device).Error; err != nil {
|
||||
return nil, errors.Wrap(errors.CodeInternalError, err, "查询设备信息失败")
|
||||
@@ -587,7 +586,6 @@ func (s *Service) validateCardOperation(ctx context.Context, deviceID, cardID ui
|
||||
return errors.New(errors.CodeDeviceNotAuthorized, "设备未授权给此企业")
|
||||
}
|
||||
|
||||
|
||||
var binding model.DeviceSimBinding
|
||||
if err := s.db.WithContext(ctx).
|
||||
Where("device_id = ? AND iot_card_id = ? AND bind_status = 1", deviceID, cardID).
|
||||
|
||||
@@ -243,9 +243,9 @@ func (s *Service) toStandaloneResponse(card *model.IotCard, shopMap map[uint]str
|
||||
CarrierName: card.CarrierName,
|
||||
IMSI: card.IMSI,
|
||||
MSISDN: card.MSISDN,
|
||||
BatchNo: card.BatchNo,
|
||||
Supplier: card.Supplier,
|
||||
Status: card.Status,
|
||||
BatchNo: card.BatchNo,
|
||||
Supplier: card.Supplier,
|
||||
Status: card.Status,
|
||||
ShopID: card.ShopID,
|
||||
ActivatedAt: card.ActivatedAt,
|
||||
ActivationStatus: card.ActivationStatus,
|
||||
|
||||
@@ -629,6 +629,7 @@ func (s *Service) CreateH5Order(ctx context.Context, req *dto.CreateOrderRequest
|
||||
purchaseRole := ""
|
||||
var sellerShopID *uint = resourceShopID
|
||||
var sellerCostPrice int64
|
||||
var expiresAt *time.Time // 待支付订单设置过期时间,立即支付的订单为 nil
|
||||
|
||||
// 场景判断:offline(平台代购)、wallet(代理钱包支付)、其他(待支付)
|
||||
if req.PaymentMethod == model.PaymentMethodOffline {
|
||||
@@ -748,6 +749,7 @@ func (s *Service) CreateH5Order(ctx context.Context, req *dto.CreateOrderRequest
|
||||
OperatorType: operatorType,
|
||||
ActualPaidAmount: actualPaidAmount,
|
||||
PurchaseRole: purchaseRole,
|
||||
ExpiresAt: expiresAt,
|
||||
}
|
||||
|
||||
items := s.buildOrderItems(userID, validationResult.Packages)
|
||||
@@ -780,6 +782,9 @@ func (s *Service) CreateH5Order(ctx context.Context, req *dto.CreateOrderRequest
|
||||
|
||||
} else {
|
||||
// 其他支付方式:创建待支付订单(H5 端支持 wechat/alipay)
|
||||
// 待支付订单设置过期时间,超过 30 分钟未支付则自动取消
|
||||
expireTime := now.Add(constants.OrderExpireTimeout)
|
||||
order.ExpiresAt = &expireTime
|
||||
if err := s.orderStore.Create(ctx, order, items); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -908,7 +913,7 @@ func (s *Service) createWalletTransaction(ctx context.Context, tx *gorm.DB, wall
|
||||
RelatedShopID: relatedShopID,
|
||||
Remark: &remark,
|
||||
Creator: userID,
|
||||
ShopIDTag: 0, // 将在下面填充
|
||||
ShopIDTag: 0, // 将在下面填充
|
||||
EnterpriseIDTag: nil,
|
||||
}
|
||||
|
||||
@@ -1144,7 +1149,124 @@ func (s *Service) Cancel(ctx context.Context, id uint, buyerType string, buyerID
|
||||
return errors.New(errors.CodeInvalidStatus, "只能取消待支付的订单")
|
||||
}
|
||||
|
||||
return s.orderStore.UpdatePaymentStatus(ctx, id, model.PaymentStatusCancelled, nil)
|
||||
return s.cancelOrder(ctx, order)
|
||||
}
|
||||
|
||||
// CancelExpiredOrders 批量取消已超时的待支付订单
|
||||
// 返回已取消的订单数量和错误
|
||||
func (s *Service) CancelExpiredOrders(ctx context.Context) (int, error) {
|
||||
startTime := time.Now()
|
||||
|
||||
orders, err := s.orderStore.FindExpiredOrders(ctx, constants.OrderExpireBatchSize)
|
||||
if err != nil {
|
||||
return 0, errors.Wrap(errors.CodeDatabaseError, err, "查询超时订单失败")
|
||||
}
|
||||
|
||||
if len(orders) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
cancelledCount := 0
|
||||
for _, order := range orders {
|
||||
if err := s.cancelOrder(ctx, order); err != nil {
|
||||
s.logger.Error("自动取消超时订单失败",
|
||||
zap.Uint("order_id", order.ID),
|
||||
zap.String("order_no", order.OrderNo),
|
||||
zap.Error(err),
|
||||
)
|
||||
continue
|
||||
}
|
||||
cancelledCount++
|
||||
}
|
||||
|
||||
s.logger.Info("批量取消超时订单完成",
|
||||
zap.Int("total", len(orders)),
|
||||
zap.Int("cancelled", cancelledCount),
|
||||
zap.Int("failed", len(orders)-cancelledCount),
|
||||
zap.Duration("duration", time.Since(startTime)),
|
||||
)
|
||||
|
||||
return cancelledCount, nil
|
||||
}
|
||||
|
||||
// cancelOrder 内部取消订单逻辑(共用于手动取消和自动超时取消)
|
||||
// 在事务中执行:更新订单状态为已取消、清除过期时间、解冻钱包余额(如有)
|
||||
func (s *Service) cancelOrder(ctx context.Context, order *model.Order) error {
|
||||
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
// 使用条件更新确保幂等性:只有待支付的订单才能取消
|
||||
result := tx.Model(&model.Order{}).
|
||||
Where("id = ? AND payment_status = ?", order.ID, model.PaymentStatusPending).
|
||||
Updates(map[string]any{
|
||||
"payment_status": model.PaymentStatusCancelled,
|
||||
"expires_at": nil,
|
||||
})
|
||||
if result.Error != nil {
|
||||
return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新订单状态失败")
|
||||
}
|
||||
if result.RowsAffected == 0 {
|
||||
// 订单已被处理(幂等),直接返回
|
||||
return nil
|
||||
}
|
||||
|
||||
// 检查是否需要解冻钱包余额(混合支付场景)
|
||||
// 当前系统中钱包支付订单是立即支付的,不会进入待支付状态
|
||||
// 此处为预留逻辑,支持未来混合支付场景的钱包解冻
|
||||
if order.PaymentMethod == model.PaymentMethodWallet {
|
||||
if err := s.unfreezeWalletForCancel(ctx, tx, order); err != nil {
|
||||
s.logger.Error("取消订单时解冻钱包失败",
|
||||
zap.Uint("order_id", order.ID),
|
||||
zap.Error(err),
|
||||
)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// unfreezeWalletForCancel 取消订单时解冻钱包余额
|
||||
// 根据买家类型和订单金额确定解冻金额和目标钱包
|
||||
func (s *Service) unfreezeWalletForCancel(ctx context.Context, tx *gorm.DB, order *model.Order) error {
|
||||
if order.BuyerType == model.BuyerTypeAgent {
|
||||
// 代理商钱包(店铺钱包)
|
||||
wallet, err := s.agentWalletStore.GetMainWallet(ctx, order.BuyerID)
|
||||
if err != nil {
|
||||
return errors.Wrap(errors.CodeWalletNotFound, err, "查询代理钱包失败")
|
||||
}
|
||||
return s.agentWalletStore.UnfreezeBalanceWithTx(ctx, tx, wallet.ID, order.TotalAmount)
|
||||
} else if order.BuyerType == model.BuyerTypePersonal {
|
||||
// 个人客户钱包(卡/设备钱包)
|
||||
var resourceType string
|
||||
var resourceID uint
|
||||
if order.OrderType == model.OrderTypeSingleCard && order.IotCardID != nil {
|
||||
resourceType = "iot_card"
|
||||
resourceID = *order.IotCardID
|
||||
} else if order.OrderType == model.OrderTypeDevice && order.DeviceID != nil {
|
||||
resourceType = "device"
|
||||
resourceID = *order.DeviceID
|
||||
} else {
|
||||
return errors.New(errors.CodeInternalError, "无法确定钱包归属")
|
||||
}
|
||||
wallet, err := s.cardWalletStore.GetByResourceTypeAndID(ctx, resourceType, resourceID)
|
||||
if err != nil {
|
||||
return errors.Wrap(errors.CodeWalletNotFound, err, "查询卡钱包失败")
|
||||
}
|
||||
// 卡钱包解冻:直接减少冻结余额
|
||||
result := tx.Model(&model.CardWallet{}).
|
||||
Where("id = ? AND frozen_balance >= ?", wallet.ID, order.TotalAmount).
|
||||
Updates(map[string]any{
|
||||
"frozen_balance": gorm.Expr("frozen_balance - ?", order.TotalAmount),
|
||||
})
|
||||
if result.Error != nil {
|
||||
return result.Error
|
||||
}
|
||||
if result.RowsAffected == 0 {
|
||||
return errors.New(errors.CodeInsufficientBalance, "冻结余额不足,无法解冻")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) WalletPay(ctx context.Context, orderID uint, buyerType string, buyerID uint) error {
|
||||
@@ -1208,6 +1330,7 @@ func (s *Service) WalletPay(ctx context.Context, orderID uint, buyerType string,
|
||||
"payment_status": model.PaymentStatusPaid,
|
||||
"payment_method": model.PaymentMethodWallet,
|
||||
"paid_at": now,
|
||||
"expires_at": nil, // 支付成功,清除过期时间
|
||||
})
|
||||
if result.Error != nil {
|
||||
return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新订单支付状态失败")
|
||||
@@ -1267,6 +1390,7 @@ func (s *Service) WalletPay(ctx context.Context, orderID uint, buyerType string,
|
||||
"payment_status": model.PaymentStatusPaid,
|
||||
"payment_method": model.PaymentMethodWallet,
|
||||
"paid_at": now,
|
||||
"expires_at": nil, // 支付成功,清除过期时间
|
||||
})
|
||||
if result.Error != nil {
|
||||
return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新订单支付状态失败")
|
||||
@@ -1332,6 +1456,7 @@ func (s *Service) HandlePaymentCallback(ctx context.Context, orderNo string, pay
|
||||
"payment_status": model.PaymentStatusPaid,
|
||||
"payment_method": paymentMethod,
|
||||
"paid_at": now,
|
||||
"expires_at": nil, // 支付成功,清除过期时间
|
||||
})
|
||||
if result.Error != nil {
|
||||
return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新订单支付状态失败")
|
||||
@@ -1844,6 +1969,10 @@ func (s *Service) buildOrderResponse(order *model.Order, items []*model.OrderIte
|
||||
Items: itemResponses,
|
||||
CreatedAt: order.CreatedAt,
|
||||
UpdatedAt: order.UpdatedAt,
|
||||
|
||||
// 订单超时信息
|
||||
ExpiresAt: order.ExpiresAt,
|
||||
IsExpired: order.ExpiresAt != nil && order.PaymentStatus == model.PaymentStatusPending && time.Now().After(*order.ExpiresAt),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -16,8 +16,8 @@ import (
|
||||
)
|
||||
|
||||
type Service struct {
|
||||
packageSeriesStore *postgres.PackageSeriesStore
|
||||
shopSeriesAllocationStore *postgres.ShopSeriesAllocationStore
|
||||
packageSeriesStore *postgres.PackageSeriesStore
|
||||
shopSeriesAllocationStore *postgres.ShopSeriesAllocationStore
|
||||
}
|
||||
|
||||
func New(packageSeriesStore *postgres.PackageSeriesStore, shopSeriesAllocationStore *postgres.ShopSeriesAllocationStore) *Service {
|
||||
|
||||
@@ -133,6 +133,16 @@ func (s *OrderStore) List(ctx context.Context, opts *store.QueryOptions, filters
|
||||
if v, ok := filters["end_time"]; ok {
|
||||
query = query.Where("created_at <= ?", v)
|
||||
}
|
||||
if v, ok := filters["is_expired"]; ok {
|
||||
isExpired, _ := v.(bool)
|
||||
if isExpired {
|
||||
// 已过期:expires_at 不为空且小于当前时间,且订单仍为待支付状态
|
||||
query = query.Where("expires_at IS NOT NULL AND expires_at <= NOW() AND payment_status = ?", model.PaymentStatusPending)
|
||||
} else {
|
||||
// 未过期:expires_at 为空或 expires_at 大于当前时间
|
||||
query = query.Where("expires_at IS NULL OR expires_at > NOW()")
|
||||
}
|
||||
}
|
||||
|
||||
if err := query.Count(&total).Error; err != nil {
|
||||
return nil, 0, err
|
||||
@@ -156,13 +166,17 @@ func (s *OrderStore) List(ctx context.Context, opts *store.QueryOptions, filters
|
||||
return orders, total, nil
|
||||
}
|
||||
|
||||
func (s *OrderStore) UpdatePaymentStatus(ctx context.Context, id uint, status int, paidAt *time.Time) error {
|
||||
func (s *OrderStore) UpdatePaymentStatus(ctx context.Context, id uint, status int, paidAt *time.Time, expiresAt ...*time.Time) error {
|
||||
updates := map[string]any{
|
||||
"payment_status": status,
|
||||
}
|
||||
if paidAt != nil {
|
||||
updates["paid_at"] = paidAt
|
||||
}
|
||||
// 支持可选的 expiresAt 参数,用于支付成功后清除过期时间或取消时清除过期时间
|
||||
if len(expiresAt) > 0 {
|
||||
updates["expires_at"] = expiresAt[0]
|
||||
}
|
||||
return s.db.WithContext(ctx).Model(&model.Order{}).Where("id = ?", id).Updates(updates).Error
|
||||
}
|
||||
|
||||
@@ -171,3 +185,19 @@ func (s *OrderStore) GenerateOrderNo() string {
|
||||
randomNum := rand.Intn(1000000)
|
||||
return fmt.Sprintf("ORD%s%06d", now.Format("20060102150405"), randomNum)
|
||||
}
|
||||
|
||||
// FindExpiredOrders 查询已超时的待支付订单
|
||||
// 查询条件:expires_at <= NOW() AND payment_status = 1(待支付)
|
||||
// limit 参数限制每次批量处理的数量,避免一次性加载太多数据
|
||||
func (s *OrderStore) FindExpiredOrders(ctx context.Context, limit int) ([]*model.Order, error) {
|
||||
var orders []*model.Order
|
||||
err := s.db.WithContext(ctx).
|
||||
Where("expires_at IS NOT NULL AND expires_at <= NOW() AND payment_status = ?", model.PaymentStatusPending).
|
||||
Order("expires_at ASC").
|
||||
Limit(limit).
|
||||
Find(&orders).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return orders, nil
|
||||
}
|
||||
|
||||
35
internal/task/alert_check.go
Normal file
35
internal/task/alert_check.go
Normal file
@@ -0,0 +1,35 @@
|
||||
package task
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/hibiken/asynq"
|
||||
"go.uber.org/zap"
|
||||
|
||||
pollingSvc "github.com/break/junhong_cmp_fiber/internal/service/polling"
|
||||
)
|
||||
|
||||
// AlertCheckHandler 告警检查任务处理器
|
||||
type AlertCheckHandler struct {
|
||||
alertService *pollingSvc.AlertService
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
||||
// NewAlertCheckHandler 创建告警检查处理器
|
||||
func NewAlertCheckHandler(alertService *pollingSvc.AlertService, logger *zap.Logger) *AlertCheckHandler {
|
||||
return &AlertCheckHandler{
|
||||
alertService: alertService,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// HandleAlertCheck 处理告警检查任务
|
||||
// 由 Asynq Scheduler 每分钟触发,检查所有告警规则
|
||||
func (h *AlertCheckHandler) HandleAlertCheck(ctx context.Context, _ *asynq.Task) error {
|
||||
if err := h.alertService.CheckAlerts(ctx); err != nil {
|
||||
h.logger.Error("告警检查失败", zap.Error(err))
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
37
internal/task/data_cleanup.go
Normal file
37
internal/task/data_cleanup.go
Normal file
@@ -0,0 +1,37 @@
|
||||
package task
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/hibiken/asynq"
|
||||
"go.uber.org/zap"
|
||||
|
||||
pollingSvc "github.com/break/junhong_cmp_fiber/internal/service/polling"
|
||||
)
|
||||
|
||||
// DataCleanupHandler 数据清理任务处理器
|
||||
type DataCleanupHandler struct {
|
||||
cleanupService *pollingSvc.CleanupService
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
||||
// NewDataCleanupHandler 创建数据清理处理器
|
||||
func NewDataCleanupHandler(cleanupService *pollingSvc.CleanupService, logger *zap.Logger) *DataCleanupHandler {
|
||||
return &DataCleanupHandler{
|
||||
cleanupService: cleanupService,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// HandleDataCleanup 处理数据清理任务
|
||||
// 由 Asynq Scheduler 每天凌晨 2 点触发,执行定期数据清理
|
||||
func (h *DataCleanupHandler) HandleDataCleanup(ctx context.Context, _ *asynq.Task) error {
|
||||
h.logger.Info("开始执行定时数据清理")
|
||||
|
||||
if err := h.cleanupService.RunScheduledCleanup(ctx); err != nil {
|
||||
h.logger.Error("定时数据清理失败", zap.Error(err))
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
43
internal/task/order_expire.go
Normal file
43
internal/task/order_expire.go
Normal file
@@ -0,0 +1,43 @@
|
||||
package task
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/hibiken/asynq"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// OrderExpirer 订单超时取消接口(局部定义,避免循环依赖)
|
||||
type OrderExpirer interface {
|
||||
CancelExpiredOrders(ctx context.Context) (int, error)
|
||||
}
|
||||
|
||||
// OrderExpireHandler 订单超时自动取消任务处理器
|
||||
type OrderExpireHandler struct {
|
||||
orderExpirer OrderExpirer
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
||||
// NewOrderExpireHandler 创建订单超时处理器
|
||||
func NewOrderExpireHandler(orderExpirer OrderExpirer, logger *zap.Logger) *OrderExpireHandler {
|
||||
return &OrderExpireHandler{
|
||||
orderExpirer: orderExpirer,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// HandleOrderExpire 处理订单超时取消任务
|
||||
// 由 Asynq Scheduler 每分钟触发,扫描并取消所有已超时的待支付订单
|
||||
func (h *OrderExpireHandler) HandleOrderExpire(ctx context.Context, _ *asynq.Task) error {
|
||||
cancelled, err := h.orderExpirer.CancelExpiredOrders(ctx)
|
||||
if err != nil {
|
||||
h.logger.Error("订单超时自动取消失败", zap.Error(err))
|
||||
return err
|
||||
}
|
||||
|
||||
if cancelled > 0 {
|
||||
h.logger.Info("订单超时自动取消完成", zap.Int("cancelled", cancelled))
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user