hd/services/pay/loop.go

package pay_service

import (
	"go.uber.org/zap"
	"hudongzhuanjia/logger"
	"hudongzhuanjia/models"
	"hudongzhuanjia/utils/define"
	"math"
	"time"
)

var orderDelayQueue = make(chan *models.UserOrder, math.MaxInt8)

func PutOrderDelayQueue(order *models.UserOrder) {
	orderDelayQueue <- order
}

func loopOrder() {
	orders, err := models.GetUserOrdersByStatus(0, 3)
	if err != nil {
		panic(err)
	}

	for _, order := range orders {
		PutOrderDelayQueue(order)
	}

	defer func() {
		if err := recover(); err != nil {
			logger.Error("订单轮询查询: panic 恢复错误", err)
		}
		// 重启
		time.Sleep(5 * time.Second)
		loopOrder()
	}()
	for {
		select {
		case order, ok := <-orderDelayQueue:
			if !ok {
				panic("通道异常关闭")
			}
			if order.ExpireAt <= time.Now().Unix() { // 订单超时
				if order.Status == 0 {
					go HandleTimeout(order)
				}
				continue
			}

			if order.Status == 0 {
				res, err := OrderQuery(order.OutTradeNo)
				if err != nil {
					logger.Error("查询订单出现错误: 错误原因-->", err.Error(), "交易订单号-->", order.OutTradeNo)
					orderDelayQueue <- order // 重新进入队列
					continue
				}

				if res["trade_state"] == define.CODE_TRADE_SUCCESS {
					go HandleSuccess(order)
				} else if res["trade_state"] == define.CODE_TRADE_REFUND {
					order.Status = 3
					orderDelayQueue <- order
					continue
				} else {
					orderDelayQueue <- order
					continue
				}
			} else if order.Status == 3 {
				res, err := QueryRefund(order.OutTradeNo)
				if err != nil {
					logger.Error("退款订单查询错误: 错误原因-->", err.Error(), "交易订单号-->", order.OutTradeNo)
					continue
				}
				order.RefundAccount = res.RefundList[0].RefundAccount
				order.RefundRecvAccount = res.RefundList[0].RefundRecvAccout
				order.Status = 4
				_, err = models.Update(order.Id, order, "refund_account", "refund_recv_account", "status")
				if err != nil {
					logger.Error("退款状态改变错误: 错误原因-->", err, "交易订单号-->", order.OutTradeNo)
					continue
				}
			}
		}
	}
}

func HandleTimeout(order *models.UserOrder) error {
	// 退还库存
	if order.GoodType == 4 {
		err := HandleCancelOrder(order.OutTradeNo)
		if err != nil {
			logger.Error(err)
		}
	}
	return nil
}

func init() {
	//go loopTransfer()
}

var transferDelayQueue = make(chan *transferDelayQueueParam, math.MaxInt8)

type transferDelayQueueParam struct {
	first          bool   // 首次跳过
	Retries        int    `json:"retries"`          // 尝试次数
	Delay          int    `json:"delay"`            // 延迟时间, 单位second
	Amount         int    `json:"amount"`           // 转账金额
	Desc           string `json:"desc"`             // 转账描述
	OpenId         string `json:"open_id"`          // 被转账人
	PartnerTradeNo string `json:"partner_trade_no"` // 转账账单单号
}

func loopTransfer() {
	//初始化
	transfers, err := models.GetUserTransferByStatus(0, 1, 3)
	if err != nil {
		panic(err)
	}

	transferDelayQueue = make(chan *transferDelayQueueParam, math.MaxInt8)
	for _, transfer := range transfers {
		transferDelayQueue <- &transferDelayQueueParam{
			first:          true,
			Retries:        5,
			Delay:          2 * 60,
			Amount:         transfer.PaymentAmount,
			Desc:           transfer.Desc,
			OpenId:         transfer.OpenId,
			PartnerTradeNo: transfer.PartnerTradeNo,
		}
	}

	defer func() {
		if errRec := recover(); errRec != nil {
			logger.Error("转账轮询 loop transfer panic", zap.Any("错误原因", errRec))
		}
		loopTransfer()
	}()
	for {
		select {
		case param, ok := <-transferDelayQueue:
			if !ok {
				panic("转账延迟通道异常关闭")
			}

			// 尝试次数
			if param.Retries <= 0 {
				logger.Info("微信转账尝试3次失败", zap.String("转账账单单号", param.PartnerTradeNo))
				userTransfer := new(models.UserTransfer)
				userTransfer.Status = 4
				_, err = userTransfer.UpdateByPartnerTradeNo(param.PartnerTradeNo)
				if err != nil {
					logger.Info("微信转账更新状态失败", zap.String("失败原因", err.Error()),
						zap.String("转账账单单号", param.PartnerTradeNo))
				}
				continue
			} else {
				param.Retries--
			}

			if !param.first {
				time.Sleep(time.Duration(param.Delay) * time.Second)
			}
			param.first = false
			res, err := TransferInfo(param.PartnerTradeNo)
			if err != nil {
				logger.Error("微信转账查询出现的错误", zap.String("错误原因", err.Error()),
					zap.String("转账账单", param.PartnerTradeNo))
				transferDelayQueue <- param
				continue
			}

			if res.Status == define.CODE_SUCCESS {
				continue
			} else if res.Status == define.CODE_TRANSFER_PROCESSING {
				transferDelayQueue <- param
				continue
			} else {
				//失败 --> 重新转账
				_, err = Transfer(param.Desc, param.OpenId, param.PartnerTradeNo, param.Amount)
				if err != nil {
					logger.Error("微信转账出现的错误", zap.String("错误原因", err.Error()),
						zap.String("转账账单", param.PartnerTradeNo))
				}
				transferDelayQueue <- param // 重新确认
				continue
			}
		}
	}
}

func PutTransferDelayQueue(desc, openId, partnerTradeNo string, amount, retries, delay int) {
	if retries <= 0 {
		retries = 3
	}
	if delay == 0 {
		delay = 30
	}
	transferDelayQueue <- &transferDelayQueueParam{
		first:          true,
		Retries:        retries,
		Delay:          delay,
		Amount:         amount,
		Desc:           desc,
		OpenId:         openId,
		PartnerTradeNo: partnerTradeNo,
	}
}