From 030425be33d74274c7af0a17d7d87b74f6936c3a Mon Sep 17 00:00:00 2001
From: huang <youdang2000@gmail.com>
Date: Fri, 27 Mar 2026 23:31:49 +0800
Subject: [PATCH] =?UTF-8?q?feat(01-03):=20AutoPurchaseHandler=20=E6=B3=A8?=
 =?UTF-8?q?=E5=85=A5=20asynqClient=EF=BC=8C=E8=A7=A6=E5=8F=91=E4=BD=A3?=
 =?UTF-8?q?=E9=87=91=E5=B9=B6=E6=B3=A8=E5=86=8C=E5=A4=84=E7=90=86=E5=99=A8?=
 =?UTF-8?q?=EF=BC=88CRITICAL-04/05=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- AutoPurchaseHandler struct 新增 asynqClient *asynq.Client 字段
- NewAutoPurchaseHandler() 新增 asynqClient 参数
- ProcessTask 事务成功后入队 TaskTypeCommission（事务外，防止一致性问题）
- pkg/queue/handler.go 新增 registerAutoPurchaseHandler() 方法
- RegisterHandlers() 调用 registerAutoPurchaseHandler()
---
 internal/task/auto_purchase.go | 25 +++++++++++++++++++++++++
 pkg/queue/handler.go           | 17 +++++++++++++++++
 2 files changed, 42 insertions(+)

diff --git a/internal/task/auto_purchase.go b/internal/task/auto_purchase.go
index 5792678..2a2f873 100644
--- a/internal/task/auto_purchase.go
+++ b/internal/task/auto_purchase.go
@@ -31,6 +31,7 @@ type AutoPurchaseHandler struct {
 	walletTransactionStore *postgres.AssetWalletTransactionStore
 	packageUsageStore      *postgres.PackageUsageStore
 	redis                  *redis.Client
+	asynqClient            *asynq.Client // 用于事务提交成功后触发佣金计算任务
 	logger                 *zap.Logger
 }
 
@@ -43,6 +44,7 @@ func NewAutoPurchaseHandler(
 	walletTransactionStore *postgres.AssetWalletTransactionStore,
 	packageUsageStore *postgres.PackageUsageStore,
 	redisClient *redis.Client,
+	asynqClient *asynq.Client,
 	logger *zap.Logger,
 ) *AutoPurchaseHandler {
 	if orderStore == nil {
@@ -69,6 +71,7 @@ func NewAutoPurchaseHandler(
 		walletTransactionStore: walletTransactionStore,
 		packageUsageStore:      packageUsageStore,
 		redis:                  redisClient,
+		asynqClient:            asynqClient,
 		logger:                 logger,
 	}
 }
@@ -121,6 +124,7 @@ func (h *AutoPurchaseHandler) ProcessTask(ctx context.Context, task *asynq.Task)
 		return err
 	}
 
+	var createdOrderID uint
 	if err := h.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
 		wallet, walletErr := h.walletStore.GetByID(ctx, rechargeRecord.AssetWalletID)
 		if walletErr != nil {
@@ -146,6 +150,7 @@ func (h *AutoPurchaseHandler) ProcessTask(ctx context.Context, task *asynq.Task)
 		if err = tx.Create(order).Error; err != nil {
 			return err
 		}
+		createdOrderID = order.ID
 
 		for _, item := range orderItems {
 			item.OrderID = order.ID
@@ -195,6 +200,26 @@ func (h *AutoPurchaseHandler) ProcessTask(ctx context.Context, task *asynq.Task)
 		return err
 	}
 
+	// 事务提交成功后触发佣金计算（不在事务内，防止任务提交后事务回滚的数据一致性问题）
+	if h.asynqClient != nil && createdOrderID > 0 {
+		payloadBytes, marshalErr := sonic.Marshal(map[string]any{"order_id": createdOrderID})
+		if marshalErr != nil {
+			h.logger.Warn("佣金任务载荷序列化失败",
+				zap.Uint("order_id", createdOrderID),
+				zap.Error(marshalErr))
+		} else {
+			commissionTask := asynq.NewTask(constants.TaskTypeCommission, payloadBytes,
+				asynq.Queue(constants.QueueLow),
+				asynq.MaxRetry(3),
+			)
+			if _, enqueueErr := h.asynqClient.EnqueueContext(ctx, commissionTask); enqueueErr != nil {
+				h.logger.Warn("自动购包后提交佣金任务失败",
+					zap.Uint("order_id", createdOrderID),
+					zap.Error(enqueueErr))
+			}
+		}
+	}
+
 	h.logger.Info("自动购包任务执行成功", zap.Uint("recharge_record_id", rechargeRecord.ID))
 	return nil
 }
diff --git a/pkg/queue/handler.go b/pkg/queue/handler.go
index 4db1162..115e64f 100644
--- a/pkg/queue/handler.go
+++ b/pkg/queue/handler.go
@@ -71,6 +71,7 @@ func (h *Handler) RegisterHandlers() *asynq.ServeMux {
 	h.registerOrderExpireHandler()
 	h.registerAlertCheckHandler()
 	h.registerDataCleanupHandler()
+	h.registerAutoPurchaseHandler()
 
 	h.logger.Info("所有任务处理器注册完成")
 	return h.mux
@@ -204,6 +205,22 @@ func (h *Handler) registerDataCleanupHandler() {
 	h.logger.Info("注册数据清理任务处理器", zap.String("task_type", constants.TaskTypeDataCleanup))
 }
 
+func (h *Handler) registerAutoPurchaseHandler() {
+	autoPurchaseHandler := task.NewAutoPurchaseHandler(
+		h.db,
+		h.workerResult.Stores.Order,
+		nil, // AssetRechargeStore：在 NewAutoPurchaseHandler 内按需初始化
+		h.workerResult.Stores.AssetWallet,
+		nil, // AssetWalletTransactionStore：在 NewAutoPurchaseHandler 内按需初始化
+		h.workerResult.Stores.PackageUsage,
+		h.redis,
+		h.asynqClient,
+		h.logger,
+	)
+	h.mux.HandleFunc(constants.TaskTypeAutoPurchaseAfterRecharge, autoPurchaseHandler.ProcessTask)
+	h.logger.Info("注册自动购包任务处理器", zap.String("task_type", constants.TaskTypeAutoPurchaseAfterRecharge))
+}
+
 // GetMux 获取 ServeMux（用于启动 Worker 服务器）
 func (h *Handler) GetMux() *asynq.ServeMux {
 	return h.mux