platform/web/tasks/task.go

package tasks

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"platform/web/events"
	m "platform/web/models"
	s "platform/web/services"

	"github.com/hibiken/asynq"
)

func HandleCompleteTrade(_ context.Context, task *asynq.Task) error {
	slog.Info("[event]尝试结束交易")
	var event events.CloseTradeData
	if err := json.Unmarshal(task.Payload(), &event); err != nil {
		return fmt.Errorf("解析任务参数失败: %w", err)
	}

	data := s.TradeRef{
		TradeNo: event.TradeNo,
		Method:  event.Method,
	}

	// 关闭交易
	err := s.Trade.CancelTrade(&data)
	if errors.Is(err, s.ErrTradeStatusIgnored) {
		result, err := s.Trade.CheckTrade(&data)
		if err != nil {
			return fmt.Errorf("检查交易状态失败: %w", err)
		}

		switch result.Status {
		case m.TradeStatusSuccess:
			if err := s.Trade.UpdateRemark(data.TradeNo, "已付款"); err != nil {
				slog.Error("添加备注失败", "err", err)
			}
		case m.TradeStatusCanceled:
			slog.Debug("交易已取消", "status", result.Status)
		default:
			return fmt.Errorf("意外交易状态: %v", result.Status)
		}

		return nil
	}

	if err != nil {
		return fmt.Errorf("结束交易失败:取消交易失败: %w", err)
	}

	return nil
}

func HandleRemoveChannel(_ context.Context, task *asynq.Task) (err error) {
	batch := string(task.Payload())
	slog.Info("[event]删除通道", "batch", batch)

	err = s.Channel.RemoveChannels(batch)
	if err != nil {
		return fmt.Errorf("删除通道失败: %w", err)
	}
	return nil
}

func HandleRefreshEdges(_ context.Context, task *asynq.Task) (err error) {
	slog.Info("[event]刷新边缘节点")

	// err = s.Channel.RefreshEdges()
	// if err != nil {
	// 	return fmt.Errorf("刷新边缘节点失败: %w", err)
	// }
	return nil
}