package queue import ( "context" "database/sql" "encoding/hex" "encoding/json" "fmt" "os" "qnc-server/app/user/cmd/api/internal/svc" "qnc-server/app/user/cmd/api/internal/types" "qnc-server/app/user/model" "qnc-server/pkg/lzkit/crypto" "qnc-server/pkg/lzkit/lzUtils" "time" "github.com/bytedance/sonic" "github.com/hibiken/asynq" "github.com/zeromicro/go-zero/core/logx" ) type PaySuccessNotifyUserHandler struct { svcCtx *svc.ServiceContext } func NewPaySuccessNotifyUserHandler(svcCtx *svc.ServiceContext) *PaySuccessNotifyUserHandler { return &PaySuccessNotifyUserHandler{ svcCtx: svcCtx, } } var payload struct { OrderID int64 `json:"order_id"` } func (l *PaySuccessNotifyUserHandler) ProcessTask(ctx context.Context, t *asynq.Task) error { // 从任务的负载中解码数据 if err := json.Unmarshal(t.Payload(), &payload); err != nil { return fmt.Errorf("解析任务负载失败: %w", err) } order, err := l.svcCtx.OrderModel.FindOne(ctx, payload.OrderID) if err != nil { return fmt.Errorf("无效的订单ID: %d, %v", payload.OrderID, err) } env := os.Getenv("ENV") if order.Status != "paid" && env != "development" { err = fmt.Errorf("无效的订单: %d", payload.OrderID) logx.Errorf("处理任务失败,原因: %v", err) return asynq.SkipRetry } product, err := l.svcCtx.ProductModel.FindOne(ctx, order.ProductId) if err != nil { return fmt.Errorf("找不到相关产品: orderID: %d, productID: %d", payload.OrderID, order.ProductId) } redisKey := fmt.Sprintf(types.QueryCacheKey, order.UserId, order.OrderNo) cache, cacheErr := l.svcCtx.Redis.GetCtx(ctx, redisKey) if cacheErr != nil { return fmt.Errorf("获取缓存内容失败: %+v", cacheErr) } var data types.QueryCacheLoad err = sonic.Unmarshal([]byte(cache), &data) if err != nil { return fmt.Errorf("解析缓存内容失败: %+v", err) } secretKey := l.svcCtx.Config.Encrypt.SecretKey key, decodeErr := hex.DecodeString(secretKey) if decodeErr != nil { return fmt.Errorf("获取AES密钥失败: %+v", decodeErr) } decryptData, aesdecryptErr := crypto.AesDecrypt(data.Params, key) if aesdecryptErr != nil { return fmt.Errorf("解密参数失败: %+v", aesdecryptErr) } // 从数据库中查询完整的查询记录 query, err := l.svcCtx.QueryModel.FindOneByOrderId(ctx, order.Id) if err != nil { return fmt.Errorf("获取插入后的查询记录失败: %+v", err) } combinedResponse, err := l.svcCtx.ApiRequestService.ProcessRequests(decryptData, product.Id) if err != nil { return l.handleError(ctx, err, order, query) } // 加密返回响应 encryptData, aesEncryptErr := crypto.AesEncrypt(combinedResponse, key) if aesEncryptErr != nil { err = fmt.Errorf("加密响应信息失败: %v", aesEncryptErr) return l.handleError(ctx, err, order, query) } query.QueryData = lzUtils.StringToNullString(encryptData) updateErr := l.svcCtx.QueryModel.UpdateWithVersion(ctx, nil, query) if updateErr != nil { err = fmt.Errorf("保存响应数据失败: %v", updateErr) return l.handleError(ctx, err, order, query) } query.QueryState = "success" updateQueryErr := l.svcCtx.QueryModel.UpdateWithVersion(ctx, nil, query) if updateQueryErr != nil { updateQueryErr = fmt.Errorf("修改查询状态失败: %v", updateQueryErr) return l.handleError(ctx, updateQueryErr, order, query) } err = l.svcCtx.AgentService.AgentProcess(ctx, order) if err != nil { return l.handleError(ctx, err, order, query) } _, delErr := l.svcCtx.Redis.DelCtx(ctx, redisKey) if delErr != nil { logx.Errorf("删除Redis缓存失败,但任务已成功处理,订单ID: %d, 错误: %v", order.Id, delErr) } return nil } // 定义一个中间件函数 func (l *PaySuccessNotifyUserHandler) handleError(ctx context.Context, err error, order *model.Order, query *model.Query) error { logx.Errorf("处理任务失败,原因: %v", err) redisKey := fmt.Sprintf(types.QueryCacheKey, order.UserId, order.OrderNo) _, delErr := l.svcCtx.Redis.DelCtx(ctx, redisKey) if delErr != nil { logx.Errorf("删除Redis缓存失败,订单ID: %d, 错误: %v", order.Id, delErr) } if order.Status == "paid" && query.QueryState == "pending" { // 更新查询状态为失败 query.QueryState = "failed" updateQueryErr := l.svcCtx.QueryModel.UpdateWithVersion(ctx, nil, query) if updateQueryErr != nil { logx.Errorf("更新查询状态失败,订单ID: %d, 错误: %v", order.Id, updateQueryErr) return asynq.SkipRetry } // 退款 if order.PaymentPlatform == "wechat" { refundErr := l.svcCtx.WechatPayService.WeChatRefund(ctx, order.OrderNo, order.Amount, order.Amount) if refundErr != nil { logx.Error(refundErr) return asynq.SkipRetry } } else { refund, refundErr := l.svcCtx.AlipayService.AliRefund(ctx, order.OrderNo, order.Amount) if refundErr != nil { logx.Error(refundErr) return asynq.SkipRetry } if refund.IsSuccess() { logx.Errorf("支付宝退款成功, orderID: %d", order.Id) // 更新订单状态为退款 order.Status = "refunded" order.RefundTime = sql.NullTime{ Time: time.Now(), Valid: true, } updateOrderErr := l.svcCtx.OrderModel.UpdateWithVersion(ctx, nil, order) if updateOrderErr != nil { logx.Errorf("更新订单状态失败,订单ID: %d, 错误: %v", order.Id, updateOrderErr) return fmt.Errorf("更新订单状态失败: %v", updateOrderErr) } return asynq.SkipRetry } else { logx.Errorf("支付宝退款失败:%v", refundErr) return asynq.SkipRetry } // 直接成功 } } return asynq.SkipRetry }