64 lines
2.1 KiB
Go
64 lines
2.1 KiB
Go
package queue
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"ycc-server/app/main/api/internal/svc"
|
||
"ycc-server/app/main/api/internal/types"
|
||
"ycc-server/app/main/model"
|
||
|
||
"github.com/hibiken/asynq"
|
||
"github.com/zeromicro/go-zero/core/logx"
|
||
)
|
||
|
||
type AgentProcessHandler struct {
|
||
svcCtx *svc.ServiceContext
|
||
}
|
||
|
||
func NewAgentProcessHandler(svcCtx *svc.ServiceContext) *AgentProcessHandler {
|
||
return &AgentProcessHandler{
|
||
svcCtx: svcCtx,
|
||
}
|
||
}
|
||
|
||
func (l *AgentProcessHandler) ProcessTask(ctx context.Context, t *asynq.Task) error {
|
||
var payload types.MsgAgentProcessPayload
|
||
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 {
|
||
if errors.Is(err, model.ErrNotFound) {
|
||
logx.Errorf("代理处理任务失败，订单不存在: orderID=%s", payload.OrderID)
|
||
return asynq.SkipRetry // 订单不存在，跳过重试
|
||
}
|
||
return fmt.Errorf("查询订单失败: orderID=%s, err=%w", payload.OrderID, err)
|
||
}
|
||
|
||
// 检查订单状态
|
||
if order.Status != "paid" {
|
||
logx.Infof("代理处理任务跳过，订单未支付: orderID=%s, status=%s", payload.OrderID, order.Status)
|
||
return nil // 订单未支付，不处理，不重试
|
||
}
|
||
|
||
// 调用代理处理服务
|
||
err = l.svcCtx.AgentService.AgentProcess(ctx, order)
|
||
if err != nil {
|
||
// 记录错误日志，但不阻塞报告流程
|
||
logx.Errorf("代理处理失败，订单ID: %s, 错误: %v", payload.OrderID, err)
|
||
// 返回错误以触发重试机制
|
||
return fmt.Errorf("代理处理失败: orderID=%s, err=%w", payload.OrderID, err)
|
||
}
|
||
|
||
// 注意：解冻任务现在通过定时任务扫描处理，不再需要发送延迟任务
|
||
// 定时任务每5分钟扫描一次待解冻的任务，更加可靠
|
||
logx.Infof("代理处理成功，订单ID: %s，冻结任务（如有）将由定时任务自动处理", payload.OrderID)
|
||
|
||
logx.Infof("代理处理成功，订单ID: %s", payload.OrderID)
|
||
return nil
|
||
}
|