team/hm-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix add

Browse Source
This commit is contained in:
18278715334
2025-12-30 17:55:00 +08:00
parent 1150c43219
commit 12659c1d14
15 changed files with 704 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
app/main/api/internal/handler/routes.go
View File

@@ -568,6 +568,14 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
rest.WithPrefix("/api/v1/admin/query"),
)
server.AddRoutes(
rest.WithMiddlewares(
[]rest.Middleware{serverCtx.AdminAuthInterceptor},
[]rest.Route{}...,
),
rest.WithPrefix("/api/v1"),
)
server.AddRoutes(
rest.WithMiddlewares(
[]rest.Middleware{serverCtx.AdminAuthInterceptor},

13
app/main/api/internal/queue/routes.go
View File

@@ -2,9 +2,9 @@ package queue
import (
"context"
"fmt"
"tydata-server/app/main/api/internal/svc"
"tydata-server/app/main/api/internal/types"
"fmt"
"github.com/hibiken/asynq"
)
@@ -24,17 +24,28 @@ func NewCronJob(ctx context.Context, svcCtx *svc.ServiceContext) *CronJob {
func (l *CronJob) Register() *asynq.ServeMux {
redisClientOpt := asynq.RedisClientOpt{Addr: l.svcCtx.Config.CacheRedis[0].Host, Password: l.svcCtx.Config.CacheRedis[0].Pass}
scheduler := asynq.NewScheduler(redisClientOpt, nil)
// 注册清理查询数据任务
task := asynq.NewTask(types.MsgCleanQueryData, nil, nil)
_, err := scheduler.Register(TASKTIME, task)
if err != nil {
panic(fmt.Sprintf("定时任务注册失败:%v", err))
}
// 注释掉原来的佣金解冻定时任务，因为现在改为按需处理单个佣金记录
// unfreezeTask := asynq.NewTask(types.MsgUnfreezeCommission, nil, nil)
// _, err = scheduler.Register("0 4 * * *", unfreezeTask)
// if err != nil {
// panic(fmt.Sprintf("佣金解冻任务注册失败:%v", err))
// }
scheduler.Start()
fmt.Println("定时任务启动！！！")
mux := asynq.NewServeMux()
mux.Handle(types.MsgPaySuccessQuery, NewPaySuccessNotifyUserHandler(l.svcCtx))
mux.Handle(types.MsgCleanQueryData, NewCleanQueryDataHandler(l.svcCtx))
mux.Handle(types.MsgUnfreezeCommission, NewUnfreezeCommissionHandler(l.svcCtx))
return mux
}

119
app/main/api/internal/queue/unfreezeCommission.go Normal file
View File

@@ -0,0 +1,119 @@
package queue
import (
"context"
"encoding/json"
"fmt"
"time"
"tydata-server/app/main/api/internal/svc"
"tydata-server/app/main/api/internal/types"
"github.com/hibiken/asynq"
"github.com/zeromicro/go-zero/core/logx"
"github.com/zeromicro/go-zero/core/stores/sqlx"
)
// 定义佣金状态常量
const (
CommissionStatusReleased = 0 // 已发放
CommissionStatusFrozen = 1 // 冻结佣金
)
// UNFREEZE_COMMISSION_DELAY_DAYS 定义延迟解冻天数
const UNFREEZE_COMMISSION_DELAY_DAYS = 3 // 三天后解冻
type UnfreezeCommissionHandler struct {
svcCtx *svc.ServiceContext
}
func NewUnfreezeCommissionHandler(svcCtx *svc.ServiceContext) *UnfreezeCommissionHandler {
return &UnfreezeCommissionHandler{
svcCtx: svcCtx,
}
}
func (l *UnfreezeCommissionHandler) ProcessTask(ctx context.Context, t *asynq.Task) error {
now := time.Now()
logx.Infof("%s - 开始执行佣金解冻任务", now.Format("2006-01-02 15:04:05"))
// 解析任务payload，获取佣金ID
var payload types.MsgUnfreezeCommissionPayload
if err := json.Unmarshal(t.Payload(), &payload); err != nil {
logx.Errorf("解析佣金解冻任务payload失败: %v", err)
return err
}
commissionID := payload.CommissionID
if commissionID <= 0 {
logx.Errorf("无效的佣金ID: %d", commissionID)
return fmt.Errorf("无效的佣金ID: %d", commissionID)
}
// 根据佣金ID查询特定佣金记录
commission, err := l.svcCtx.AgentCommissionModel.FindOne(ctx, commissionID)
if err != nil {
logx.Errorf("查询佣金记录ID %d 失败: %v", commissionID, err)
return err
}
// 检查佣金状态是否为冻结状态
if commission.Status != CommissionStatusFrozen {
logx.Infof("佣金记录ID %d 状态不是冻结状态，当前状态: %d，无需处理", commissionID, commission.Status)
return nil
}
// 检查是否已到达解冻时间（3天后）
unfreezeTime := commission.UpdateTime.AddDate(0, 0, UNFREEZE_COMMISSION_DELAY_DAYS)
if now.Before(unfreezeTime) {
logx.Infof("佣金记录ID %d 尚未到达解冻时间，解冻时间: %s", commissionID, unfreezeTime.Format("2006-01-02 15:04:05"))
return nil
}
// 使用事务处理解冻操作
err = l.svcCtx.AgentCommissionModel.Trans(ctx, func(ctx context.Context, session sqlx.Session) error {
// 更新佣金状态为已发放
commission.Status = CommissionStatusReleased
commission.UpdateTime = now
// 更新佣金数据库
_, err := l.svcCtx.AgentCommissionModel.Update(ctx, session, commission)
if err != nil {
logx.Errorf("更新佣金记录ID %d 失败: %v", commissionID, err)
return err
}
// 获取代理钱包记录
agentWallet, err := l.svcCtx.AgentWalletModel.FindOneByAgentId(ctx, commission.AgentId)
if err != nil {
logx.Errorf("查询代理ID %d 的钱包记录失败: %v", commission.AgentId, err)
return err
}
// 更新钱包余额：增加佣金金额到balance，减少相应的frozen_balance
// 注意：这里应该转移的是当前佣金记录的金额，而不是全部冻结余额
commissionAmount := commission.Amount
agentWallet.Balance += commissionAmount
agentWallet.FrozenBalance -= commissionAmount
agentWallet.UpdateTime = now
// 更新钱包数据库
_, err = l.svcCtx.AgentWalletModel.Update(ctx, session, agentWallet)
if err != nil {
logx.Errorf("更新代理ID %d 的钱包记录失败: %v", commission.AgentId, err)
return err
}
logx.Infof("成功解冻佣金记录ID %d，代理ID %d，佣金金额 %.2f，已将佣金金额从冻结余额转移到可用余额",
commissionID, commission.AgentId, commissionAmount)
return nil
})
if err != nil {
logx.Errorf("%s - 佣金解冻任务失败: %v", now.Format("2006-01-02 15:04:05"), err)
return err
}
logx.Infof("%s - 佣金解冻任务完成，佣金ID: %d", now.Format("2006-01-02 15:04:05"), commissionID)
return nil
}

84
app/main/api/internal/service/agentService.go
View File

@@ -2,11 +2,14 @@ package service
import (
"context"
"database/sql"
"tydata-server/app/main/api/internal/config"
"tydata-server/app/main/model"
"tydata-server/common/globalkey"
"tydata-server/pkg/lzkit/lzUtils"
"github.com/pkg/errors"
"github.com/zeromicro/go-zero/core/logx"
"github.com/zeromicro/go-zero/core/stores/sqlx"
)
@@ -29,6 +32,7 @@ type AgentService struct {
AgentPlatformDeductionModel model.AgentPlatformDeductionModel
AgentActiveStatModel model.AgentActiveStatModel
AgentWithdrawalModel model.AgentWithdrawalModel
AsynqService *AsynqService
}
func NewAgentService(c config.Config, orderModel model.OrderModel, agentModel model.AgentModel, agentAuditModel model.AgentAuditModel,
@@ -38,7 +42,7 @@ func NewAgentService(c config.Config, orderModel model.OrderModel, agentModel mo
agentMembershipRechargeOrderModel model.AgentMembershipRechargeOrderModel,
agentMembershipUserConfigModel model.AgentMembershipUserConfigModel,
agentProductConfigModel model.AgentProductConfigModel, agentPlatformDeductionModel model.AgentPlatformDeductionModel,
agentActiveStatModel model.AgentActiveStatModel, agentWithdrawalModel model.AgentWithdrawalModel) *AgentService {
agentActiveStatModel model.AgentActiveStatModel, agentWithdrawalModel model.AgentWithdrawalModel, asynqService *AsynqService) *AgentService {
return &AgentService{
config: c,
@@ -59,6 +63,7 @@ func NewAgentService(c config.Config, orderModel model.OrderModel, agentModel mo
AgentPlatformDeductionModel: agentPlatformDeductionModel,
AgentActiveStatModel: agentActiveStatModel,
AgentWithdrawalModel: agentWithdrawalModel,
AsynqService: asynqService,
}
}
@@ -113,34 +118,49 @@ func (l *AgentService) AgentProcess(ctx context.Context, order *model.Order) err
return findAgentMembersipConfigModelErr
}
// 定价
commissionCost, commissionCostErr := l.CommissionCost(transCtx, agentID, AncestorId, AgentMembershipConfigModel, order.ProductId, session)
commissionCost, commissionCostErr := l.CommissionCost(transCtx, agentID, AncestorId, AgentMembershipConfigModel, order.ProductId, order.Id, session)
if commissionCostErr != nil {
return commissionCostErr
}
// 提价
commissionPricing, commissionPricingErr := l.CommissionPricing(transCtx, agentID, AncestorId, AgentMembershipConfigModel, order.ProductId, order.Amount, session)
commissionPricing, commissionPricingErr := l.CommissionPricing(transCtx, agentID, AncestorId, AgentMembershipConfigModel, order.ProductId, order.Amount, order.Id, session)
if commissionPricingErr != nil {
return commissionPricingErr
}
// 上级克扣的成本
descendantDeductedAmount = commissionCost + commissionPricing
// 佣金
// 奖励
ancestorCommissionReward, ancestorCommissionErr := l.AncestorCommission(transCtx, agentID, AncestorId, session)
if ancestorCommissionErr != nil {
return ancestorCommissionErr
}
// 给上级成本以及佣金
ancestorCommissionAmount := commissionCost + commissionPricing + ancestorCommissionReward
ancestorCommissionAmount := commissionCost + commissionPricing
ancestorWallet, findAgentWalletModelErr := l.AgentWalletModel.FindOneByAgentId(transCtx, AncestorId)
if findAgentWalletModelErr != nil {
return findAgentWalletModelErr
}
// 奖励不冻结
ancestorWallet.Balance += ancestorCommissionReward
ancestorWallet.Balance += ancestorCommissionAmount
ancestorWallet.TotalEarnings += ancestorCommissionAmount
// 冻结
ancestorWallet.FrozenBalance += ancestorCommissionAmount
// 为上级创建佣金记录（冻结金额）
ancestorCommissionRecord := model.AgentCommission{
AgentId: AncestorId,
OrderId: order.Id,
Amount: ancestorCommissionAmount,
ProductId: order.ProductId,
}
_, insertAncestorCommissionErr := l.AgentCommissionModel.Insert(transCtx, session, &ancestorCommissionRecord)
if insertAncestorCommissionErr != nil {
return insertAncestorCommissionErr
}
ancestorWallet.TotalEarnings += ancestorCommissionAmount + ancestorCommissionReward
updateErr := l.AgentWalletModel.UpdateWithVersion(transCtx, session, ancestorWallet)
if updateErr != nil {
return updateErr
@@ -161,6 +181,37 @@ func (l *AgentService) AgentProcess(ctx context.Context, order *model.Order) err
return transErr
}
// 在事务提交后，触发解冻任务（3天后自动解冻）
// 注意：这里发送的是任务，实际解冻将在3天后由队列处理
if l.AsynqService != nil {
// 获取刚创建的佣金记录ID
// 由于我们需要佣金记录ID来触发解冻任务，但事务中无法获取，我们可以在事务后查询
builder := l.AgentCommissionModel.SelectBuilder().
Where("order_id = ?", order.Id).
Where("del_state = ?", globalkey.DelStateNo)
commissions, findErr := l.AgentCommissionModel.FindAll(ctx, builder, "")
if findErr != nil {
logx.Errorf("查询刚创建的佣金记录失败，订单ID: %d, 错误: %v", order.Id, findErr)
return findErr
}
if len(commissions) > 0 {
// 为所有新创建的佣金记录触发解冻任务
for _, commission := range commissions {
// 发送解冻任务，将在3天后执行
sendTaskErr := l.AsynqService.SendUnfreezeCommissionTask(commission.Id)
if sendTaskErr != nil {
logx.Errorf("发送佣金解冻任务失败，佣金ID: %d, 错误: %v", commission.Id, sendTaskErr)
// 不返回错误，因为佣金记录已创建成功，只是解冻任务失败
} else {
logx.Infof("已发送佣金解冻任务，佣金ID: %d, 代理ID: %d, 金额: %.2f",
commission.Id, commission.AgentId, commission.Amount)
}
}
}
}
return nil
}
@@ -172,7 +223,7 @@ func (l *AgentService) AgentCommission(ctx context.Context, agentID int64, order
}
// 推广人最终获得代理佣金
finalCommission := order.Amount - deductedAmount
agentWalletModel.Balance += finalCommission
agentWalletModel.FrozenBalance += finalCommission
agentWalletModel.TotalEarnings += finalCommission
agentCommission := model.AgentCommission{
@@ -181,11 +232,18 @@ func (l *AgentService) AgentCommission(ctx context.Context, agentID int64, order
Amount: finalCommission,
ProductId: order.ProductId,
}
_, insertAgentCommissionErr := l.AgentCommissionModel.Insert(ctx, session, &agentCommission)
insertResult, insertAgentCommissionErr := l.AgentCommissionModel.Insert(ctx, session, &agentCommission)
if insertAgentCommissionErr != nil {
return insertAgentCommissionErr
}
// 获取新插入的佣金记录ID（用于日志记录）
commissionID, err := insertResult.LastInsertId()
if err != nil {
return err
}
_ = commissionID // 暂时忽略该变量，因为我们使用其他方式获取佣金记录
updateAgentWalletErr := l.AgentWalletModel.UpdateWithVersion(ctx, session, agentWalletModel)
if updateAgentWalletErr != nil {
return updateAgentWalletErr
@@ -270,7 +328,7 @@ func (l *AgentService) PlatformPricing(ctx context.Context, agentID int64, order
}
// CommissionCost 上级底价成本
func (l *AgentService) CommissionCost(ctx context.Context, descendantId int64, AncestorId int64, agentMembershipConfigModel *model.AgentMembershipConfig, productID int64, session sqlx.Session) (float64, error) {
func (l *AgentService) CommissionCost(ctx context.Context, descendantId int64, AncestorId int64, agentMembershipConfigModel *model.AgentMembershipConfig, productID int64, orderId int64, session sqlx.Session) (float64, error) {
if agentMembershipConfigModel.PriceIncreaseAmount.Valid {
// 拥有则查看该上级设定的成本
agentMembershipUserConfigModel, findAgentMembershipUserConfigModelErr := l.AgentMembershipUserConfigModel.FindOneByAgentIdProductId(ctx, AncestorId, productID)
@@ -290,6 +348,7 @@ func (l *AgentService) CommissionCost(ctx context.Context, descendantId int64, A
Amount: deductCostAmount,
Type: model.AgentDeductionTypeCost,
ProductId: productID,
OrderId: sql.NullInt64{Int64: orderId, Valid: true},
}
_, insertAgentCommissionDeductionModelErr := l.AgentCommissionDeductionModel.Insert(ctx, session, &agentCommissionDeductionModel)
@@ -303,7 +362,7 @@ func (l *AgentService) CommissionCost(ctx context.Context, descendantId int64, A
}
// CommissionPricing 上级提价成本
func (l *AgentService) CommissionPricing(ctx context.Context, descendantId int64, AncestorId int64, agentMembershipConfigModel *model.AgentMembershipConfig, productID int64, pricing float64, session sqlx.Session) (float64, error) {
func (l *AgentService) CommissionPricing(ctx context.Context, descendantId int64, AncestorId int64, agentMembershipConfigModel *model.AgentMembershipConfig, productID int64, pricing float64, orderId int64, session sqlx.Session) (float64, error) {
//看上级代理等级否有拥有定价标准收益功能
if agentMembershipConfigModel.PriceIncreaseMax.Valid && agentMembershipConfigModel.PriceRatio.Valid {
// 拥有则查看该上级设定的成本
@@ -334,6 +393,7 @@ func (l *AgentService) CommissionPricing(ctx context.Context, descendantId int64
Amount: deductCostAmount,
Type: model.AgentDeductionTypePricing,
ProductId: productID,
OrderId: sql.NullInt64{Int64: orderId, Valid: true},
}
_, insertAgentCommissionDeductionModelErr := l.AgentCommissionDeductionModel.Insert(ctx, session, &agentCommissionDeductionModel)
if insertAgentCommissionDeductionModelErr != nil {

31
app/main/api/internal/service/asynqService.go
View File

@@ -3,9 +3,9 @@
package service
import (
"encoding/json"
"tydata-server/app/main/api/internal/config"
"tydata-server/app/main/api/internal/types"
"encoding/json"
"github.com/hibiken/asynq"
"github.com/zeromicro/go-zero/core/logx"
@@ -58,3 +58,32 @@ func (s *AsynqService) SendQueryTask(orderID int64) error {
logx.Infof("发送异步任务成功，任务ID: %s, 队列: %s, 订单号: %d", info.ID, info.Queue, orderID)
return nil
}
// SendUnfreezeCommissionTask 发送佣金解冻任务
func (s *AsynqService) SendUnfreezeCommissionTask(commissionID int64) error {
// 准备任务的 payload
payload := types.MsgUnfreezeCommissionPayload{
CommissionID: commissionID,
}
payloadBytes, err := json.Marshal(payload)
if err != nil {
logx.Errorf("发送佣金解冻任务失败 (无法编码 payload): %v, 佣金ID: %d", err, commissionID)
return err
}
options := []asynq.Option{
asynq.MaxRetry(5), // 设置最大重试次数
}
task := asynq.NewTask(types.MsgUnfreezeCommission, payloadBytes, options...)
// 将任务加入队列并获取任务信息
info, err := s.client.Enqueue(task)
if err != nil {
logx.Errorf("发送佣金解冻任务失败 (加入队列失败): %+v, 佣金ID: %d", err, commissionID)
return err
}
// 记录成功日志，带上任务 ID 和队列信息
logx.Infof("发送佣金解冻任务成功，任务ID: %s, 队列: %s, 佣金ID: %d", info.ID, info.Queue, commissionID)
return nil
}

2
app/main/api/internal/svc/servicecontext.go
View File

@@ -194,7 +194,7 @@ func NewServiceContext(c config.Config) *ServiceContext {
agentCommissionModel, agentCommissionDeductionModel, agentWalletModel, agentLinkModel,
agentOrderModel, agentRewardsModel, agentMembershipConfigModel, agentMembershipRechargeOrderModel,
agentMembershipUserConfigModel, agentProductConfigModel, agentPlatformDeductionModel,
agentActiveStatModel, agentWithdrawalModel)
agentActiveStatModel, agentWithdrawalModel, asynqService)
userService := service.NewUserService(&c, userModel, userAuthModel, userTempModel, agentModel)
dictService := service.NewDictService(adminDictTypeModel, adminDictDataModel)
adminPromotionLinkStatsService := service.NewAdminPromotionLinkStatsService(adminPromotionLinkModel,

4
app/main/api/internal/types/payload.go
View File

@@ -3,3 +3,7 @@ package types
type MsgPaySuccessQueryPayload struct {
OrderID int64 `json:"order_id"`
}
type MsgUnfreezeCommissionPayload struct {
CommissionID int64 `json:"commission_id"`
}

1
app/main/api/internal/types/taskname.go
View File

@@ -2,3 +2,4 @@ package types
const MsgPaySuccessQuery = "msg:pay_success:query"
const MsgCleanQueryData = "msg:clean_query_data"
const MsgUnfreezeCommission = "msg:unfreeze_commission"