qnc-server-tob/app/main/api/internal/queue/paySuccessNotify.go

package queue

import (
	"context"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"os"
	"qnc-server/app/main/api/internal/svc"
	"qnc-server/app/main/api/internal/types"
	"qnc-server/app/main/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 := &model.Query{
		OrderId:     order.Id,
		UserId:      order.UserId,
		ProductId:   product.Id,
		QueryParams: data.Params,
		QueryState:  "pending",
	}
	result, insertQueryErr := l.svcCtx.QueryModel.Insert(ctx, nil, query)
	if insertQueryErr != nil {
		return fmt.Errorf("保存查询失败: %+v", insertQueryErr)
	}

	// 获取插入后的ID
	queryId, err := result.LastInsertId()
	if err != nil {
		return fmt.Errorf("获取插入的查询ID失败: %+v", err)
	}
	// 从数据库中查询完整的查询记录
	query, err = l.svcCtx.QueryModel.FindOne(ctx, queryId)
	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
}