csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity

feat: 钱包系统分离 - 代理钱包与卡钱包完全隔离
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 7m17s
Details

Browse Source 
## 变更概述
将统一钱包系统拆分为代理钱包和卡钱包两个独立系统，实现数据表和代码层面的完全隔离。

## 数据库变更
- 新增 6 张表：tb_agent_wallet、tb_agent_wallet_transaction、tb_agent_recharge_record、tb_card_wallet、tb_card_wallet_transaction、tb_card_recharge_record
- 删除 3 张旧表：tb_wallet、tb_wallet_transaction、tb_recharge_record
- 代理钱包：按 (shop_id, wallet_type) 唯一标识，支持主钱包和分佣钱包
- 卡钱包：按 (resource_type, resource_id) 唯一标识，支持物联网卡和设备

## 代码变更
- Model 层：新增 AgentWallet、AgentWalletTransaction、AgentRechargeRecord、CardWallet、CardWalletTransaction、CardRechargeRecord 模型
- Store 层：新增 6 个独立 Store，支持事务、乐观锁、Redis 缓存
- Service 层：重构 commission_calculation、commission_withdrawal、order、recharge 等 8 个服务
- Bootstrap 层：更新 Store 和 Service 依赖注入
- 常量层：按钱包类型重新组织常量和 Redis Key 生成函数

## 技术特性
- 乐观锁：使用 version 字段防止并发冲突
- 多租户：支持 shop_id_tag 和 enterprise_id_tag 过滤
- 事务管理：所有余额变动使用事务保证 ACID
- 缓存策略：Cache-Aside 模式，余额变动后删除缓存

## 业务影响
- 代理钱包和卡钱包业务完全隔离，互不影响
- 为独立监控、优化、扩展打下基础
- 提升代理钱包的稳定性和独立性

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
huang
2026-02-25 09:51:00 +08:00
parent f32d32cd36
commit 18daeae65a
66 changed files with 4420 additions and 1090 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
internal/service/commission_calculation/service.go
View File

@@ -14,22 +14,22 @@ import (
)
type Service struct {
db *gorm.DB
commissionRecordStore *postgres.CommissionRecordStore
shopStore *postgres.ShopStore
shopPackageAllocationStore *postgres.ShopPackageAllocationStore
shopSeriesAllocationStore *postgres.ShopSeriesAllocationStore
packageSeriesStore *postgres.PackageSeriesStore
iotCardStore *postgres.IotCardStore
deviceStore *postgres.DeviceStore
walletStore *postgres.WalletStore
walletTransactionStore *postgres.WalletTransactionStore
orderStore *postgres.OrderStore
orderItemStore *postgres.OrderItemStore
packageStore *postgres.PackageStore
commissionStatsStore *postgres.ShopSeriesCommissionStatsStore
commissionStatsService *commission_stats.Service
logger *zap.Logger
db *gorm.DB
commissionRecordStore *postgres.CommissionRecordStore
shopStore *postgres.ShopStore
shopPackageAllocationStore *postgres.ShopPackageAllocationStore
shopSeriesAllocationStore *postgres.ShopSeriesAllocationStore
packageSeriesStore *postgres.PackageSeriesStore
iotCardStore *postgres.IotCardStore
deviceStore *postgres.DeviceStore
agentWalletStore *postgres.AgentWalletStore
agentWalletTransactionStore *postgres.AgentWalletTransactionStore
orderStore *postgres.OrderStore
orderItemStore *postgres.OrderItemStore
packageStore *postgres.PackageStore
commissionStatsStore *postgres.ShopSeriesCommissionStatsStore
commissionStatsService *commission_stats.Service
logger *zap.Logger
}
func New(
@@ -41,8 +41,8 @@ func New(
packageSeriesStore *postgres.PackageSeriesStore,
iotCardStore *postgres.IotCardStore,
deviceStore *postgres.DeviceStore,
walletStore *postgres.WalletStore,
walletTransactionStore *postgres.WalletTransactionStore,
agentWalletStore *postgres.AgentWalletStore,
agentWalletTransactionStore *postgres.AgentWalletTransactionStore,
orderStore *postgres.OrderStore,
orderItemStore *postgres.OrderItemStore,
packageStore *postgres.PackageStore,
@@ -51,22 +51,22 @@ func New(
logger *zap.Logger,
) *Service {
return &Service{
db: db,
commissionRecordStore: commissionRecordStore,
shopStore: shopStore,
shopPackageAllocationStore: shopPackageAllocationStore,
shopSeriesAllocationStore: shopSeriesAllocationStore,
packageSeriesStore: packageSeriesStore,
iotCardStore: iotCardStore,
deviceStore: deviceStore,
walletStore: walletStore,
walletTransactionStore: walletTransactionStore,
orderStore: orderStore,
orderItemStore: orderItemStore,
packageStore: packageStore,
commissionStatsStore: commissionStatsStore,
commissionStatsService: commissionStatsService,
logger: logger,
db: db,
commissionRecordStore: commissionRecordStore,
shopStore: shopStore,
shopPackageAllocationStore: shopPackageAllocationStore,
shopSeriesAllocationStore: shopSeriesAllocationStore,
packageSeriesStore: packageSeriesStore,
iotCardStore: iotCardStore,
deviceStore: deviceStore,
agentWalletStore: agentWalletStore,
agentWalletTransactionStore: agentWalletTransactionStore,
orderStore: orderStore,
orderItemStore: orderItemStore,
packageStore: packageStore,
commissionStatsStore: commissionStatsStore,
commissionStatsService: commissionStatsService,
logger: logger,
}
}
@@ -562,25 +562,20 @@ func (s *Service) matchOneTimeCommissionTier(ctx context.Context, shopID uint, s
}
func (s *Service) creditCommissionInTx(ctx context.Context, tx *gorm.DB, record *model.CommissionRecord) error {
var wallet model.Wallet
if err := tx.Where("resource_type = ? AND resource_id = ? AND wallet_type = ?", "shop", record.ShopID, "commission").First(&wallet).Error; err != nil {
return errors.Wrap(errors.CodeDatabaseError, err, "获取店铺钱包失败")
// 获取店铺的分佣钱包
var wallet model.AgentWallet
if err := tx.Where("shop_id = ? AND wallet_type = ?", record.ShopID, "commission").First(&wallet).Error; err != nil {
return errors.Wrap(errors.CodeDatabaseError, err, "获取店铺分佣钱包失败")
}
balanceBefore := wallet.Balance
result := tx.Model(&model.Wallet{}).
Where("id = ? AND version = ?", wallet.ID, wallet.Version).
Updates(map[string]any{
"balance": gorm.Expr("balance + ?", record.Amount),
"version": gorm.Expr("version + 1"),
})
if result.Error != nil {
return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新钱包余额失败")
}
if result.RowsAffected == 0 {
return errors.New(errors.CodeInternalError, "钱包版本冲突，请重试")
// 使用 AgentWalletStore 的方法增加余额（带事务）
if err := s.agentWalletStore.AddBalanceWithTx(ctx, tx, wallet.ID, record.Amount); err != nil {
return errors.Wrap(errors.CodeInternalError, err, "更新钱包余额失败")
}
// 更新佣金记录状态
now := time.Now()
if err := tx.Model(record).Updates(map[string]any{
"balance_after": balanceBefore + record.Amount,
@@ -590,9 +585,11 @@ func (s *Service) creditCommissionInTx(ctx context.Context, tx *gorm.DB, record
return errors.Wrap(errors.CodeDatabaseError, err, "更新佣金记录失败")
}
// 创建代理钱包交易记录
remark := "佣金入账"
transaction := &model.WalletTransaction{
WalletID: wallet.ID,
transaction := &model.AgentWalletTransaction{
AgentWalletID: wallet.ID,
ShopID: record.ShopID,
UserID: record.Creator,
TransactionType: "commission",
Amount: record.Amount,
@@ -603,8 +600,9 @@ func (s *Service) creditCommissionInTx(ctx context.Context, tx *gorm.DB, record
ReferenceID: &record.ID,
Remark: &remark,
Creator: record.Creator,
ShopIDTag: record.ShopID,
}
if err := tx.Create(transaction).Error; err != nil {
if err := s.agentWalletTransactionStore.CreateWithTx(ctx, tx, transaction); err != nil {
return errors.Wrap(errors.CodeDatabaseError, err, "创建钱包交易记录失败")
}

39
internal/service/commission_withdrawal/service.go
View File

@@ -19,8 +19,8 @@ type Service struct {
db *gorm.DB
shopStore *postgres.ShopStore
accountStore *postgres.AccountStore
walletStore *postgres.WalletStore
walletTransactionStore *postgres.WalletTransactionStore
agentWalletStore *postgres.AgentWalletStore
agentWalletTransactionStore *postgres.AgentWalletTransactionStore
commissionWithdrawalReqStore *postgres.CommissionWithdrawalRequestStore
}
@@ -28,16 +28,16 @@ func New(
db *gorm.DB,
shopStore *postgres.ShopStore,
accountStore *postgres.AccountStore,
walletStore *postgres.WalletStore,
walletTransactionStore *postgres.WalletTransactionStore,
agentWalletStore *postgres.AgentWalletStore,
agentWalletTransactionStore *postgres.AgentWalletTransactionStore,
commissionWithdrawalReqStore *postgres.CommissionWithdrawalRequestStore,
) *Service {
return &Service{
db: db,
shopStore: shopStore,
accountStore: accountStore,
walletStore: walletStore,
walletTransactionStore: walletTransactionStore,
agentWalletStore: agentWalletStore,
agentWalletTransactionStore: agentWalletTransactionStore,
commissionWithdrawalReqStore: commissionWithdrawalReqStore,
}
}
@@ -157,7 +157,8 @@ func (s *Service) Approve(ctx context.Context, id uint, req *dto.ApproveWithdraw
return nil, errors.New(errors.CodeInvalidStatus, "申请状态不允许此操作")
}
wallet, err := s.walletStore.GetShopCommissionWallet(ctx, withdrawal.ShopID)
// 获取店铺分佣钱包
wallet, err := s.agentWalletStore.GetCommissionWallet(ctx, withdrawal.ShopID)
if err != nil {
return nil, errors.New(errors.CodeNotFound, "店铺佣金钱包不存在")
}
@@ -173,14 +174,17 @@ func (s *Service) Approve(ctx context.Context, id uint, req *dto.ApproveWithdraw
now := time.Now()
err = s.db.Transaction(func(tx *gorm.DB) error {
if err := s.walletStore.DeductFrozenBalanceWithTx(ctx, tx, wallet.ID, amount); err != nil {
// 从冻结余额扣款
if err := s.agentWalletStore.DeductFrozenBalanceWithTx(ctx, tx, wallet.ID, amount); err != nil {
return errors.Wrap(errors.CodeInternalError, err, "扣除冻结余额失败")
}
// 创建代理钱包交易记录
refType := "withdrawal"
refID := withdrawal.ID
transaction := &model.WalletTransaction{
WalletID: wallet.ID,
transaction := &model.AgentWalletTransaction{
AgentWalletID: wallet.ID,
ShopID: withdrawal.ShopID,
UserID: currentUserID,
TransactionType: "withdrawal",
Amount: -amount,
@@ -190,8 +194,9 @@ func (s *Service) Approve(ctx context.Context, id uint, req *dto.ApproveWithdraw
ReferenceType: &refType,
ReferenceID: &refID,
Creator: currentUserID,
ShopIDTag: withdrawal.ShopID,
}
if err := s.walletTransactionStore.CreateWithTx(ctx, tx, transaction); err != nil {
if err := s.agentWalletTransactionStore.CreateWithTx(ctx, tx, transaction); err != nil {
return errors.Wrap(errors.CodeInternalError, err, "创建交易流水失败")
}
@@ -265,21 +270,22 @@ func (s *Service) Reject(ctx context.Context, id uint, req *dto.RejectWithdrawal
return nil, errors.New(errors.CodeInvalidStatus, "申请状态不允许此操作")
}
wallet, err := s.walletStore.GetShopCommissionWallet(ctx, withdrawal.ShopID)
wallet, err := s.agentWalletStore.GetCommissionWallet(ctx, withdrawal.ShopID)
if err != nil {
return nil, errors.New(errors.CodeNotFound, "店铺佣金钱包不存在")
}
now := time.Now()
err = s.db.Transaction(func(tx *gorm.DB) error {
if err := s.walletStore.UnfreezeBalanceWithTx(ctx, tx, wallet.ID, withdrawal.Amount); err != nil {
if err := s.agentWalletStore.UnfreezeBalanceWithTx(ctx, tx, wallet.ID, withdrawal.Amount); err != nil {
return errors.Wrap(errors.CodeInternalError, err, "解冻余额失败")
}
refType := "withdrawal"
refID := withdrawal.ID
transaction := &model.WalletTransaction{
WalletID: wallet.ID,
transaction := &model.AgentWalletTransaction{
AgentWalletID: wallet.ID,
ShopID: withdrawal.ShopID,
UserID: currentUserID,
TransactionType: "refund",
Amount: withdrawal.Amount,
@@ -289,8 +295,9 @@ func (s *Service) Reject(ctx context.Context, id uint, req *dto.RejectWithdrawal
ReferenceType: &refType,
ReferenceID: &refID,
Creator: currentUserID,
ShopIDTag: withdrawal.ShopID,
}
if err := s.walletTransactionStore.CreateWithTx(ctx, tx, transaction); err != nil {
if err := s.agentWalletTransactionStore.CreateWithTx(ctx, tx, transaction); err != nil {
return errors.Wrap(errors.CodeInternalError, err, "创建交易流水失败")
}

2
internal/service/iot_card/stop_resume_service.go
View File

@@ -42,7 +42,7 @@ func NewStopResumeService(
iotCardStore: iotCardStore,
gatewayClient: gatewayClient,
logger: logger,
maxRetries: 3, // 默认最多重试 3 次
maxRetries: 3, // 默认最多重试 3 次
retryInterval: 2 * time.Second, // 默认重试间隔 2 秒
}
}

26
internal/service/my_commission/service.go
View File

@@ -19,30 +19,30 @@ import (
type Service struct {
db *gorm.DB
shopStore *postgres.ShopStore
walletStore *postgres.WalletStore
agentWalletStore *postgres.AgentWalletStore
commissionWithdrawalRequestStore *postgres.CommissionWithdrawalRequestStore
commissionWithdrawalSettingStore *postgres.CommissionWithdrawalSettingStore
commissionRecordStore *postgres.CommissionRecordStore
walletTransactionStore *postgres.WalletTransactionStore
agentWalletTransactionStore *postgres.AgentWalletTransactionStore
}
func New(
db *gorm.DB,
shopStore *postgres.ShopStore,
walletStore *postgres.WalletStore,
agentWalletStore *postgres.AgentWalletStore,
commissionWithdrawalRequestStore *postgres.CommissionWithdrawalRequestStore,
commissionWithdrawalSettingStore *postgres.CommissionWithdrawalSettingStore,
commissionRecordStore *postgres.CommissionRecordStore,
walletTransactionStore *postgres.WalletTransactionStore,
agentWalletTransactionStore *postgres.AgentWalletTransactionStore,
) *Service {
return &Service{
db: db,
shopStore: shopStore,
walletStore: walletStore,
agentWalletStore: agentWalletStore,
commissionWithdrawalRequestStore: commissionWithdrawalRequestStore,
commissionWithdrawalSettingStore: commissionWithdrawalSettingStore,
commissionRecordStore: commissionRecordStore,
walletTransactionStore: walletTransactionStore,
agentWalletTransactionStore: agentWalletTransactionStore,
}
}
@@ -63,8 +63,8 @@ func (s *Service) GetCommissionSummary(ctx context.Context) (*dto.MyCommissionSu
return nil, errors.New(errors.CodeShopNotFound, "店铺不存在")
}
// 使用 GetShopCommissionWallet 获取店铺佣金钱包
wallet, err := s.walletStore.GetShopCommissionWallet(ctx, shopID)
// 使用 GetCommissionWallet 获取店铺佣金钱包
wallet, err := s.agentWalletStore.GetCommissionWallet(ctx, shopID)
if err != nil {
// 钱包不存在时返回空数据
return &dto.MyCommissionSummaryResp{
@@ -120,7 +120,7 @@ func (s *Service) CreateWithdrawalRequest(ctx context.Context, req *dto.CreateMy
}
// 获取钱包
wallet, err := s.walletStore.GetShopCommissionWallet(ctx, shopID)
wallet, err := s.agentWalletStore.GetCommissionWallet(ctx, shopID)
if err != nil {
return nil, errors.New(errors.CodeInsufficientBalance, "钱包不存在")
}
@@ -160,7 +160,7 @@ func (s *Service) CreateWithdrawalRequest(ctx context.Context, req *dto.CreateMy
err = s.db.Transaction(func(tx *gorm.DB) error {
// 冻结余额
if err := tx.WithContext(ctx).Model(&model.Wallet{}).
if err := tx.WithContext(ctx).Model(&model.AgentWallet{}).
Where("id = ? AND balance >= ?", wallet.ID, req.Amount).
Updates(map[string]interface{}{
"balance": gorm.Expr("balance - ?", req.Amount),
@@ -192,8 +192,9 @@ func (s *Service) CreateWithdrawalRequest(ctx context.Context, req *dto.CreateMy
// 创建钱包流水记录
remark := fmt.Sprintf("提现冻结，单号：%s", withdrawalNo)
refType := constants.ReferenceTypeWithdrawal
transaction := &model.WalletTransaction{
WalletID: wallet.ID,
transaction := &model.AgentWalletTransaction{
AgentWalletID: wallet.ID,
ShopID: shopID,
UserID: currentUserID,
TransactionType: constants.TransactionTypeWithdrawal,
Amount: -req.Amount, // 冻结为负数
@@ -204,6 +205,7 @@ func (s *Service) CreateWithdrawalRequest(ctx context.Context, req *dto.CreateMy
ReferenceID: &withdrawalRequest.ID,
Remark: &remark,
Creator: currentUserID,
ShopIDTag: shopID,
}
if err := tx.WithContext(ctx).Create(transaction).Error; err != nil {

169
internal/service/order/service.go
View File

@@ -30,7 +30,8 @@ type Service struct {
redis *redis.Client
orderStore *postgres.OrderStore
orderItemStore *postgres.OrderItemStore
walletStore *postgres.WalletStore
agentWalletStore *postgres.AgentWalletStore
cardWalletStore *postgres.CardWalletStore
purchaseValidationService *purchase_validation.Service
shopPackageAllocationStore *postgres.ShopPackageAllocationStore
shopSeriesAllocationStore *postgres.ShopSeriesAllocationStore
@@ -49,7 +50,8 @@ func New(
redisClient *redis.Client,
orderStore *postgres.OrderStore,
orderItemStore *postgres.OrderItemStore,
walletStore *postgres.WalletStore,
agentWalletStore *postgres.AgentWalletStore,
cardWalletStore *postgres.CardWalletStore,
purchaseValidationService *purchase_validation.Service,
shopPackageAllocationStore *postgres.ShopPackageAllocationStore,
shopSeriesAllocationStore *postgres.ShopSeriesAllocationStore,
@@ -67,7 +69,8 @@ func New(
redis: redisClient,
orderStore: orderStore,
orderItemStore: orderItemStore,
walletStore: walletStore,
agentWalletStore: agentWalletStore,
cardWalletStore: cardWalletStore,
purchaseValidationService: purchaseValidationService,
shopPackageAllocationStore: shopPackageAllocationStore,
shopSeriesAllocationStore: shopSeriesAllocationStore,
@@ -430,64 +433,128 @@ func (s *Service) WalletPay(ctx context.Context, orderID uint, buyerType string,
return errors.New(errors.CodeInvalidParam, "不支持的买家类型")
}
wallet, err := s.walletStore.GetByResourceTypeAndID(ctx, resourceType, resourceID, "main")
if err != nil {
if err == gorm.ErrRecordNotFound {
return errors.New(errors.CodeWalletNotFound, "钱包不存在")
}
return err
}
if wallet.Balance < order.TotalAmount {
return errors.New(errors.CodeInsufficientBalance, "余额不足")
}
// 根据资源类型选择对应的钱包系统
now := time.Now()
err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
result := tx.Model(&model.Order{}).
Where("id = ? AND payment_status = ?", orderID, model.PaymentStatusPending).
Updates(map[string]any{
"payment_status": model.PaymentStatusPaid,
"payment_method": model.PaymentMethodWallet,
"paid_at": now,
})
if result.Error != nil {
return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新订单支付状态失败")
if resourceType == "shop" {
// 代理钱包系统（店铺）
wallet, err := s.agentWalletStore.GetMainWallet(ctx, resourceID)
if err != nil {
if err == gorm.ErrRecordNotFound {
return errors.New(errors.CodeWalletNotFound, "钱包不存在")
}
return err
}
if result.RowsAffected == 0 {
var currentOrder model.Order
if err := tx.First(&currentOrder, orderID).Error; err != nil {
return errors.Wrap(errors.CodeDatabaseError, err, "查询订单失败")
if wallet.Balance < order.TotalAmount {
return errors.New(errors.CodeInsufficientBalance, "余额不足")
}
err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
result := tx.Model(&model.Order{}).
Where("id = ? AND payment_status = ?", orderID, model.PaymentStatusPending).
Updates(map[string]any{
"payment_status": model.PaymentStatusPaid,
"payment_method": model.PaymentMethodWallet,
"paid_at": now,
})
if result.Error != nil {
return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新订单支付状态失败")
}
switch currentOrder.PaymentStatus {
case model.PaymentStatusPaid:
return nil
case model.PaymentStatusCancelled:
return errors.New(errors.CodeInvalidStatus, "订单已取消，无法支付")
case model.PaymentStatusRefunded:
return errors.New(errors.CodeInvalidStatus, "订单已退款，无法支付")
default:
return errors.New(errors.CodeInvalidStatus, "订单状态异常")
if result.RowsAffected == 0 {
var currentOrder model.Order
if err := tx.First(&currentOrder, orderID).Error; err != nil {
return errors.Wrap(errors.CodeDatabaseError, err, "查询订单失败")
}
switch currentOrder.PaymentStatus {
case model.PaymentStatusPaid:
return nil
case model.PaymentStatusCancelled:
return errors.New(errors.CodeInvalidStatus, "订单已取消，无法支付")
case model.PaymentStatusRefunded:
return errors.New(errors.CodeInvalidStatus, "订单已退款，无法支付")
default:
return errors.New(errors.CodeInvalidStatus, "订单状态异常")
}
}
walletResult := tx.Model(&model.AgentWallet{}).
Where("id = ? AND balance >= ? AND version = ?", wallet.ID, order.TotalAmount, wallet.Version).
Updates(map[string]any{
"balance": gorm.Expr("balance - ?", order.TotalAmount),
"version": gorm.Expr("version + 1"),
})
if walletResult.Error != nil {
return errors.Wrap(errors.CodeDatabaseError, walletResult.Error, "扣减钱包余额失败")
}
if walletResult.RowsAffected == 0 {
return errors.New(errors.CodeInsufficientBalance, "余额不足或并发冲突")
}
return s.activatePackage(ctx, tx, order)
})
} else {
// 卡钱包系统（iot_card 或 device）
wallet, err := s.cardWalletStore.GetByResourceTypeAndID(ctx, resourceType, resourceID)
if err != nil {
if err == gorm.ErrRecordNotFound {
return errors.New(errors.CodeWalletNotFound, "钱包不存在")
}
return err
}
walletResult := tx.Model(&model.Wallet{}).
Where("id = ? AND balance >= ? AND version = ?", wallet.ID, order.TotalAmount, wallet.Version).
Updates(map[string]any{
"balance": gorm.Expr("balance - ?", order.TotalAmount),
"version": gorm.Expr("version + 1"),
})
if walletResult.Error != nil {
return errors.Wrap(errors.CodeDatabaseError, walletResult.Error, "扣减钱包余额失败")
}
if walletResult.RowsAffected == 0 {
return errors.New(errors.CodeInsufficientBalance, "余额不足或并发冲突")
if wallet.Balance < order.TotalAmount {
return errors.New(errors.CodeInsufficientBalance, "余额不足")
}
return s.activatePackage(ctx, tx, order)
})
err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
result := tx.Model(&model.Order{}).
Where("id = ? AND payment_status = ?", orderID, model.PaymentStatusPending).
Updates(map[string]any{
"payment_status": model.PaymentStatusPaid,
"payment_method": model.PaymentMethodWallet,
"paid_at": now,
})
if result.Error != nil {
return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新订单支付状态失败")
}
if result.RowsAffected == 0 {
var currentOrder model.Order
if err := tx.First(&currentOrder, orderID).Error; err != nil {
return errors.Wrap(errors.CodeDatabaseError, err, "查询订单失败")
}
switch currentOrder.PaymentStatus {
case model.PaymentStatusPaid:
return nil
case model.PaymentStatusCancelled:
return errors.New(errors.CodeInvalidStatus, "订单已取消，无法支付")
case model.PaymentStatusRefunded:
return errors.New(errors.CodeInvalidStatus, "订单已退款，无法支付")
default:
return errors.New(errors.CodeInvalidStatus, "订单状态异常")
}
}
walletResult := tx.Model(&model.CardWallet{}).
Where("id = ? AND balance >= ? AND version = ?", wallet.ID, order.TotalAmount, wallet.Version).
Updates(map[string]any{
"balance": gorm.Expr("balance - ?", order.TotalAmount),
"version": gorm.Expr("version + 1"),
})
if walletResult.Error != nil {
return errors.Wrap(errors.CodeDatabaseError, walletResult.Error, "扣减钱包余额失败")
}
if walletResult.RowsAffected == 0 {
return errors.New(errors.CodeInsufficientBalance, "余额不足或并发冲突")
}
return s.activatePackage(ctx, tx, order)
})
}
if err != nil {
return err

9
internal/service/package/reset_service.go
View File

@@ -71,10 +71,10 @@ func (s *ResetService) resetDailyUsageWithDB(ctx context.Context, db *gorm.DB) e
// 批量更新
updates := map[string]interface{}{
"data_usage_mb": 0,
"last_reset_at": now,
"next_reset_at": nextReset,
"status": constants.PackageUsageStatusActive, // 重置后恢复为生效中
"data_usage_mb": 0,
"last_reset_at": now,
"next_reset_at": nextReset,
"status": constants.PackageUsageStatusActive, // 重置后恢复为生效中
}
if err := tx.Model(&model.PackageUsage{}).
@@ -221,4 +221,3 @@ func (s *ResetService) resetYearlyUsageWithDB(ctx context.Context, db *gorm.DB)
return nil
})
}

14
internal/service/polling/cleanup_service.go
View File

@@ -152,13 +152,13 @@ func (s *CleanupService) Preview(ctx context.Context) ([]*CleanupPreview, error)
// CleanupProgress 清理进度
type CleanupProgress struct {
IsRunning bool `json:"is_running"`
CurrentTable string `json:"current_table,omitempty"`
TotalTables int `json:"total_tables"`
ProcessedTables int `json:"processed_tables"`
TotalDeleted int64 `json:"total_deleted"`
StartedAt *time.Time `json:"started_at,omitempty"`
LastLog *model.DataCleanupLog `json:"last_log,omitempty"`
IsRunning bool `json:"is_running"`
CurrentTable string `json:"current_table,omitempty"`
TotalTables int `json:"total_tables"`
ProcessedTables int `json:"processed_tables"`
TotalDeleted int64 `json:"total_deleted"`
StartedAt *time.Time `json:"started_at,omitempty"`
LastLog *model.DataCleanupLog `json:"last_log,omitempty"`
}
// GetProgress 获取清理进度

10
internal/service/polling/concurrency_service.go
View File

@@ -28,11 +28,11 @@ func NewConcurrencyService(store *postgres.PollingConcurrencyConfigStore, redis
// ConcurrencyStatus 并发状态
type ConcurrencyStatus struct {
TaskType string `json:"task_type"`
TaskTypeName string `json:"task_type_name"`
MaxConcurrency int `json:"max_concurrency"`
Current int64 `json:"current"`
Available int64 `json:"available"`
TaskType string `json:"task_type"`
TaskTypeName string `json:"task_type_name"`
MaxConcurrency int `json:"max_concurrency"`
Current int64 `json:"current"`
Available int64 `json:"available"`
Utilization float64 `json:"utilization"`
}

14
internal/service/polling/manual_trigger_service.go
View File

@@ -207,13 +207,13 @@ func (s *ManualTriggerService) processBatchTrigger(ctx context.Context, logID ui
// ConditionFilter 条件筛选参数
type ConditionFilter struct {
CardStatus string `json:"card_status,omitempty"` // 卡状态
CarrierCode string `json:"carrier_code,omitempty"` // 运营商代码
CardType string `json:"card_type,omitempty"` // 卡类型
ShopID *uint `json:"shop_id,omitempty"` // 店铺ID
PackageIDs []uint `json:"package_ids,omitempty"` // 套餐ID列表
EnablePolling *bool `json:"enable_polling,omitempty"` // 是否启用轮询
Limit int `json:"limit,omitempty"` // 限制数量
CardStatus string `json:"card_status,omitempty"` // 卡状态
CarrierCode string `json:"carrier_code,omitempty"` // 运营商代码
CardType string `json:"card_type,omitempty"` // 卡类型
ShopID *uint `json:"shop_id,omitempty"` // 店铺ID
PackageIDs []uint `json:"package_ids,omitempty"` // 套餐ID列表
EnablePolling *bool `json:"enable_polling,omitempty"` // 是否启用轮询
Limit int `json:"limit,omitempty"` // 限制数量
}
// TriggerByCondition 条件筛选触发

139
internal/service/recharge/service.go
View File

@@ -29,26 +29,26 @@ type ForceRechargeRequirement struct {
}
// Service 充值服务
// 负责充值订单的创建、预检、支付回调处理等业务逻辑
// 负责卡钱包（IoT卡/设备）的充值订单创建、预检、支付回调处理等业务逻辑
type Service struct {
db *gorm.DB
rechargeStore *postgres.RechargeStore
walletStore *postgres.WalletStore
walletTransactionStore *postgres.WalletTransactionStore
iotCardStore *postgres.IotCardStore
deviceStore *postgres.DeviceStore
shopSeriesAllocationStore *postgres.ShopSeriesAllocationStore
packageSeriesStore *postgres.PackageSeriesStore
commissionRecordStore *postgres.CommissionRecordStore
logger *zap.Logger
db *gorm.DB
cardRechargeStore *postgres.CardRechargeStore
cardWalletStore *postgres.CardWalletStore
cardWalletTransactionStore *postgres.CardWalletTransactionStore
iotCardStore *postgres.IotCardStore
deviceStore *postgres.DeviceStore
shopSeriesAllocationStore *postgres.ShopSeriesAllocationStore
packageSeriesStore *postgres.PackageSeriesStore
commissionRecordStore *postgres.CommissionRecordStore
logger *zap.Logger
}
// New 创建充值服务实例
func New(
db *gorm.DB,
rechargeStore *postgres.RechargeStore,
walletStore *postgres.WalletStore,
walletTransactionStore *postgres.WalletTransactionStore,
cardRechargeStore *postgres.CardRechargeStore,
cardWalletStore *postgres.CardWalletStore,
cardWalletTransactionStore *postgres.CardWalletTransactionStore,
iotCardStore *postgres.IotCardStore,
deviceStore *postgres.DeviceStore,
shopSeriesAllocationStore *postgres.ShopSeriesAllocationStore,
@@ -57,16 +57,16 @@ func New(
logger *zap.Logger,
) *Service {
return &Service{
db: db,
rechargeStore: rechargeStore,
walletStore: walletStore,
walletTransactionStore: walletTransactionStore,
iotCardStore: iotCardStore,
deviceStore: deviceStore,
shopSeriesAllocationStore: shopSeriesAllocationStore,
packageSeriesStore: packageSeriesStore,
commissionRecordStore: commissionRecordStore,
logger: logger,
db: db,
cardRechargeStore: cardRechargeStore,
cardWalletStore: cardWalletStore,
cardWalletTransactionStore: cardWalletTransactionStore,
iotCardStore: iotCardStore,
deviceStore: deviceStore,
shopSeriesAllocationStore: shopSeriesAllocationStore,
packageSeriesStore: packageSeriesStore,
commissionRecordStore: commissionRecordStore,
logger: logger,
}
}
@@ -81,14 +81,14 @@ func (s *Service) Create(ctx context.Context, req *dto.CreateRechargeRequest, us
return nil, errors.New(errors.CodeRechargeAmountInvalid, "充值金额不能超过100000元")
}
// 2. 获取资源（卡或设备）
var wallet *model.Wallet
// 2. 获取资源（卡或设备）钱包
var wallet *model.CardWallet
var err error
if req.ResourceType == "iot_card" {
wallet, err = s.walletStore.GetByResourceTypeAndID(ctx, "iot_card", req.ResourceID, "main")
wallet, err = s.cardWalletStore.GetByResourceTypeAndID(ctx, "iot_card", req.ResourceID)
} else if req.ResourceType == "device" {
wallet, err = s.walletStore.GetByResourceTypeAndID(ctx, "device", req.ResourceID, "main")
wallet, err = s.cardWalletStore.GetByResourceTypeAndID(ctx, "device", req.ResourceID)
} else {
return nil, errors.New(errors.CodeInvalidParam, "无效的资源类型")
}
@@ -115,20 +115,20 @@ func (s *Service) Create(ctx context.Context, req *dto.CreateRechargeRequest, us
rechargeNo := s.generateRechargeNo()
// 5. 创建充值订单
recharge := &model.RechargeRecord{
BaseModel: model.BaseModel{
Creator: userID,
Updater: userID,
},
UserID: userID,
WalletID: wallet.ID,
RechargeNo: rechargeNo,
Amount: req.Amount,
PaymentMethod: req.PaymentMethod,
Status: constants.RechargeStatusPending,
recharge := &model.CardRechargeRecord{
UserID: userID,
CardWalletID: wallet.ID,
ResourceType: req.ResourceType,
ResourceID: req.ResourceID,
RechargeNo: rechargeNo,
Amount: req.Amount,
PaymentMethod: req.PaymentMethod,
Status: constants.RechargeStatusPending,
ShopIDTag: wallet.ShopIDTag,
EnterpriseIDTag: wallet.EnterpriseIDTag,
}
if err := s.rechargeStore.Create(ctx, recharge); err != nil {
if err := s.cardRechargeStore.Create(ctx, recharge); err != nil {
return nil, errors.Wrap(errors.CodeDatabaseError, err, "创建充值订单失败")
}
@@ -173,7 +173,7 @@ func (s *Service) GetRechargeCheck(ctx context.Context, resourceType string, res
// GetByID 根据ID查询充值订单详情
// 支持数据权限过滤
func (s *Service) GetByID(ctx context.Context, id uint, userID uint) (*dto.RechargeResponse, error) {
recharge, err := s.rechargeStore.GetByID(ctx, id)
recharge, err := s.cardRechargeStore.GetByID(ctx, id)
if err != nil {
if err == gorm.ErrRecordNotFound {
return nil, errors.New(errors.CodeRechargeNotFound, "充值订单不存在")
@@ -201,7 +201,7 @@ func (s *Service) List(ctx context.Context, req *dto.RechargeListRequest, userID
pageSize = constants.DefaultPageSize
}
params := &postgres.ListRechargeParams{
params := &postgres.ListCardRechargeParams{
Page: page,
PageSize: pageSize,
UserID: &userID, // 数据权限：只能查看自己的
@@ -211,7 +211,8 @@ func (s *Service) List(ctx context.Context, req *dto.RechargeListRequest, userID
params.Status = req.Status
}
if req.WalletID != nil {
params.WalletID = req.WalletID
walletID := *req.WalletID
params.CardWalletID = &walletID
}
if req.StartTime != nil {
params.StartTime = req.StartTime
@@ -220,7 +221,7 @@ func (s *Service) List(ctx context.Context, req *dto.RechargeListRequest, userID
params.EndTime = req.EndTime
}
recharges, total, err := s.rechargeStore.List(ctx, params)
recharges, total, err := s.cardRechargeStore.List(ctx, params)
if err != nil {
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询充值订单列表失败")
}
@@ -248,13 +249,13 @@ func (s *Service) List(ctx context.Context, req *dto.RechargeListRequest, userID
// 支持幂等性检查、事务处理、更新余额、触发佣金
func (s *Service) HandlePaymentCallback(ctx context.Context, rechargeNo string, paymentMethod string, paymentTransactionID string) error {
// 1. 查询充值订单
recharge, err := s.rechargeStore.GetByRechargeNo(ctx, rechargeNo)
recharge, err := s.cardRechargeStore.GetByRechargeNo(ctx, rechargeNo)
if err != nil {
if err == gorm.ErrRecordNotFound {
return errors.New(errors.CodeRechargeNotFound, "充值订单不存在")
}
return errors.Wrap(errors.CodeDatabaseError, err, "查询充值订单失败")
}
if recharge == nil {
return errors.New(errors.CodeRechargeNotFound, "充值订单不存在")
}
// 2. 幂等性检查：已支付则直接返回成功
if recharge.Status == constants.RechargeStatusPaid || recharge.Status == constants.RechargeStatusCompleted {
@@ -271,21 +272,21 @@ func (s *Service) HandlePaymentCallback(ctx context.Context, rechargeNo string,
}
// 4. 获取钱包信息
wallet, err := s.walletStore.GetByID(ctx, recharge.WalletID)
wallet, err := s.cardWalletStore.GetByID(ctx, recharge.CardWalletID)
if err != nil {
return errors.Wrap(errors.CodeDatabaseError, err, "查询钱包失败")
}
// 5. 获取钱包对应的资源类型和ID
resourceType := wallet.ResourceType
resourceID := wallet.ResourceID
// 5. 获取钱包对应的资源类型和ID（从充值记录中直接获取）
resourceType := recharge.ResourceType
resourceID := recharge.ResourceID
// 6. 事务处理：更新订单状态、增加余额、更新累计充值、触发佣金
now := time.Now()
err = s.db.Transaction(func(tx *gorm.DB) error {
// 6.1 更新充值订单状态（带状态检查，实现乐观锁）
oldStatus := constants.RechargeStatusPending
if err := s.rechargeStore.UpdateStatus(ctx, recharge.ID, &oldStatus, constants.RechargeStatusPaid, &now, nil); err != nil {
if err := s.cardRechargeStore.UpdateStatusWithOptimisticLock(ctx, recharge.ID, &oldStatus, constants.RechargeStatusPaid, &now, nil); err != nil {
if err == gorm.ErrRecordNotFound {
// 状态已变更，幂等处理
return nil
@@ -294,13 +295,13 @@ func (s *Service) HandlePaymentCallback(ctx context.Context, rechargeNo string,
}
// 6.2 更新支付信息
if err := s.rechargeStore.UpdatePaymentInfo(ctx, recharge.ID, &paymentMethod, &paymentTransactionID); err != nil {
if err := s.cardRechargeStore.UpdatePaymentInfo(ctx, recharge.ID, &paymentMethod, &paymentTransactionID); err != nil {
return errors.Wrap(errors.CodeDatabaseError, err, "更新支付信息失败")
}
// 6.3 增加钱包余额（使用乐观锁）
balanceBefore := wallet.Balance
result := tx.Model(&model.Wallet{}).
result := tx.Model(&model.CardWallet{}).
Where("id = ? AND version = ?", wallet.ID, wallet.Version).
Updates(map[string]any{
"balance": gorm.Expr("balance + ?", recharge.Amount),
@@ -316,8 +317,10 @@ func (s *Service) HandlePaymentCallback(ctx context.Context, rechargeNo string,
// 6.4 创建钱包交易记录
remark := "钱包充值"
refType := "recharge"
transaction := &model.WalletTransaction{
WalletID: wallet.ID,
transaction := &model.CardWalletTransaction{
CardWalletID: wallet.ID,
ResourceType: resourceType,
ResourceID: resourceID,
UserID: recharge.UserID,
TransactionType: "recharge",
Amount: recharge.Amount,
@@ -327,7 +330,8 @@ func (s *Service) HandlePaymentCallback(ctx context.Context, rechargeNo string,
ReferenceType: &refType,
ReferenceID: &recharge.ID,
Remark: &remark,
Creator: recharge.UserID,
ShopIDTag: wallet.ShopIDTag,
EnterpriseIDTag: wallet.EnterpriseIDTag,
}
if err := tx.Create(transaction).Error; err != nil {
return errors.Wrap(errors.CodeDatabaseError, err, "创建钱包交易记录失败")
@@ -344,7 +348,7 @@ func (s *Service) HandlePaymentCallback(ctx context.Context, rechargeNo string,
}
// 6.7 更新充值订单状态为已完成
if err := tx.Model(&model.RechargeRecord{}).
if err := tx.Model(&model.CardRechargeRecord{}).
Where("id = ?", recharge.ID).
Update("status", constants.RechargeStatusCompleted).Error; err != nil {
return errors.Wrap(errors.CodeDatabaseError, err, "更新充值订单完成状态失败")
@@ -590,8 +594,8 @@ func (s *Service) triggerOneTimeCommissionIfNeededInTx(ctx context.Context, tx *
return nil
}
var commissionWallet model.Wallet
if err := tx.Where("resource_type = ? AND resource_id = ? AND wallet_type = ?", "shop", *shopID, "commission").
var commissionWallet model.AgentWallet
if err := tx.Where("shop_id = ? AND wallet_type = ?", *shopID, constants.AgentWalletTypeCommission).
First(&commissionWallet).Error; err != nil {
if err == gorm.ErrRecordNotFound {
s.logger.Warn("店铺佣金钱包不存在，跳过佣金发放",
@@ -629,7 +633,7 @@ func (s *Service) triggerOneTimeCommissionIfNeededInTx(ctx context.Context, tx *
// 11. 佣金入账到店铺佣金钱包
balanceBefore := commissionWallet.Balance
result := tx.Model(&model.Wallet{}).
result := tx.Model(&model.AgentWallet{}).
Where("id = ? AND version = ?", commissionWallet.ID, commissionWallet.Version).
Updates(map[string]any{
"balance": gorm.Expr("balance + ?", commissionAmount),
@@ -654,8 +658,9 @@ func (s *Service) triggerOneTimeCommissionIfNeededInTx(ctx context.Context, tx *
// 13. 创建佣金钱包交易记录
remark := "一次性佣金入账(充值触发)"
refType := "commission"
commissionTransaction := &model.WalletTransaction{
WalletID: commissionWallet.ID,
commissionTransaction := &model.AgentWalletTransaction{
AgentWalletID: commissionWallet.ID,
ShopID: *shopID,
UserID: userID,
TransactionType: "commission",
Amount: commissionAmount,
@@ -665,7 +670,7 @@ func (s *Service) triggerOneTimeCommissionIfNeededInTx(ctx context.Context, tx *
ReferenceType: &refType,
ReferenceID: &commissionRecord.ID,
Remark: &remark,
Creator: userID,
ShopIDTag: *shopID,
}
if err := tx.Create(commissionTransaction).Error; err != nil {
return errors.Wrap(errors.CodeDatabaseError, err, "创建佣金钱包交易记录失败")
@@ -724,7 +729,7 @@ func (s *Service) generateRechargeNo() string {
}
// buildRechargeResponse 构建充值订单响应
func (s *Service) buildRechargeResponse(recharge *model.RechargeRecord) *dto.RechargeResponse {
func (s *Service) buildRechargeResponse(recharge *model.CardRechargeRecord) *dto.RechargeResponse {
statusText := ""
switch recharge.Status {
case constants.RechargeStatusPending:
@@ -743,7 +748,7 @@ func (s *Service) buildRechargeResponse(recharge *model.RechargeRecord) *dto.Rec
ID: recharge.ID,
RechargeNo: recharge.RechargeNo,
UserID: recharge.UserID,
WalletID: recharge.WalletID,
WalletID: recharge.CardWalletID,
Amount: recharge.Amount,
PaymentMethod: recharge.PaymentMethod,
PaymentChannel: recharge.PaymentChannel,

10
internal/service/shop_commission/service.go
View File

@@ -16,7 +16,7 @@ import (
type Service struct {
shopStore *postgres.ShopStore
accountStore *postgres.AccountStore
walletStore *postgres.WalletStore
agentWalletStore *postgres.AgentWalletStore
commissionWithdrawalReqStore *postgres.CommissionWithdrawalRequestStore
commissionRecordStore *postgres.CommissionRecordStore
}
@@ -24,14 +24,14 @@ type Service struct {
func New(
shopStore *postgres.ShopStore,
accountStore *postgres.AccountStore,
walletStore *postgres.WalletStore,
agentWalletStore *postgres.AgentWalletStore,
commissionWithdrawalReqStore *postgres.CommissionWithdrawalRequestStore,
commissionRecordStore *postgres.CommissionRecordStore,
) *Service {
return &Service{
shopStore: shopStore,
accountStore: accountStore,
walletStore: walletStore,
agentWalletStore: agentWalletStore,
commissionWithdrawalReqStore: commissionWithdrawalReqStore,
commissionRecordStore: commissionRecordStore,
}
@@ -74,7 +74,7 @@ func (s *Service) ListShopCommissionSummary(ctx context.Context, req *dto.ShopCo
shopIDs = append(shopIDs, shop.ID)
}
walletSummaries, err := s.walletStore.GetShopCommissionSummaryBatch(ctx, shopIDs)
walletSummaries, err := s.agentWalletStore.GetShopCommissionSummaryBatch(ctx, shopIDs)
if err != nil {
return nil, errors.Wrap(errors.CodeInternalError, err, "查询店铺钱包汇总失败")
}
@@ -123,7 +123,7 @@ func (s *Service) ListShopCommissionSummary(ctx context.Context, req *dto.ShopCo
}, nil
}
func (s *Service) buildCommissionSummaryItem(shop *model.Shop, walletSummary *postgres.ShopCommissionSummary, withdrawnAmount, withdrawingAmount int64, account *model.Account) dto.ShopCommissionSummaryItem {
func (s *Service) buildCommissionSummaryItem(shop *model.Shop, walletSummary *model.AgentWallet, withdrawnAmount, withdrawingAmount int64, account *model.Account) dto.ShopCommissionSummaryItem {
var balance, frozenBalance int64
if walletSummary != nil {
balance = walletSummary.Balance