const API = require("../../lib/API.js") const db = require("../../plugin/DataBase/db.js") const Redis = require('../../plugin/DataBase/Redis') const { BaseStdResponse } = require("../../BaseStdResponse.js") const AccessControl = require("../../lib/AccessControl.js") const crypto = require('crypto') const axios = require('axios') const config = require('../../config.json') const mq = require('../../plugin/mq') const ORDER_PAYMENT_QUEUE = 'order_payment_check' let orderPaymentWorkerStarted = false async function startOrderPaymentWorker(logger) { if (orderPaymentWorkerStarted) { return } try { const ch = await mq.getChannel('order_payment') await ch.assertQueue(ORDER_PAYMENT_QUEUE, { durable: true }) await ch.prefetch(1) logger.info(`订单支付结果轮询消费者已启动,队列:${ORDER_PAYMENT_QUEUE}`) orderPaymentWorkerStarted = true ch.consume(ORDER_PAYMENT_QUEUE, async (msg) => { if (!msg) return const content = JSON.parse(msg.content.toString() || '{}') const { orderId } = content if (!orderId) { logger.warn('收到无效的订单支付检查消息(缺少 orderId)') ch.ack(msg) return } try { await pollOrderPaymentStatus(orderId, logger) ch.ack(msg) } catch (err) { logger.error(`订单支付轮询处理失败,订单号:${orderId},错误:${err.stack || err}`) // 简单重试策略:出现异常时,Nack 并重新入队 ch.nack(msg, false, true) } }) } catch (e) { logger.error(`启动订单支付 MQ 消费者失败:${e.stack || e}`) } } async function pollOrderPaymentStatus(orderId, logger) { const paymentConfig = config.pay || {} if (!paymentConfig.pid || !paymentConfig.url || !paymentConfig.key) { logger.error('支付配置错误,无法轮询易支付状态') return } const MAX_RETRIES = 120 // 5分钟 / 5秒 const DELAY = 2500 // 5秒 const queryUrl = `${paymentConfig.url}/api.php?act=order&pid=${paymentConfig.pid}&key=${paymentConfig.key}&out_trade_no=${orderId}` const pollOrderStatus = async (retry = 0) => { if (retry >= MAX_RETRIES) { logger.info(`订单超时未支付,自动取消,订单号:${orderId}`) await db.query('UPDATE orders SET state = 3 WHERE orderId = ?', [orderId]) return } try { const queryRes = await axios.get(queryUrl) const queryData = queryRes.data console.log(`轮询订单支付状态,订单号:${orderId},尝试次数:${retry + 1},查询结果:${JSON.stringify(queryData)}`) if (queryData.code == 1 && queryData.status == 1) { const { trade_no, out_trade_no, type } = queryData const time = Date.now() let sql = 'UPDATE orders SET state = 1, pay_type = ?, pay_id = ?, pay_time = ? WHERE orderId = ? AND state = 0' const result = await db.query(sql, [type, trade_no, time, out_trade_no]) if (result.affectedRows > 0) { sql = ` SELECT g.lepao_count, g.ic_count, g.vip, a.create_user FROM orders a LEFT JOIN goods g ON a.goods_id = g.id WHERE a.orderId = ? ` const rows = await db.query(sql, [out_trade_no]) if (!rows || rows.length !== 1) { logger.error(`订单商品信息异常,订单号:${out_trade_no}`) await db.query('UPDATE orders SET state = 4 WHERE orderId = ?', [out_trade_no]) return } const { lepao_count, ic_count, vip, create_user } = rows[0] sql = 'UPDATE users SET lepao_count = lepao_count + ?, ic_count = ic_count + ?, vip = ? WHERE uuid = ?' const updateUser = await db.query(sql, [lepao_count, ic_count, vip, create_user]) if (!updateUser || updateUser.affectedRows !== 1) { logger.error(`更新用户失败,UUID: ${create_user}`) await db.query('UPDATE orders SET state = 4 WHERE orderId = ?', [out_trade_no]) } sql = 'UPDATE orders SET state = 2 WHERE orderId = ?' await db.query(sql, [out_trade_no]) logger.info(`订单处理成功:${out_trade_no}`) return } else { logger.warn(`订单不存在或已处理:${out_trade_no}`) } } // 未支付,继续轮询 setTimeout(() => pollOrderStatus(retry + 1), DELAY) } catch (error) { logger.warn(`轮询支付状态失败:${error.stack || error}`) setTimeout(() => pollOrderStatus(retry + 1), DELAY) } } logger.info(`开始轮询订单支付状态,订单号:${orderId}`) pollOrderStatus() } function generateOrderId() { const now = new Date() const pad = (n, w = 2) => n.toString().padStart(w, '0') return `${now.getFullYear()}${pad(now.getMonth() + 1)}${pad(now.getDate())}` + `${pad(now.getHours())}${pad(now.getMinutes())}${pad(now.getSeconds())}` + `${pad(now.getMilliseconds(), 3)}` } function generatePaymentSign(params, key) { const sorted = Object.keys(params).sort() const query = sorted.map(k => `${k}=${params[k]}`).join('&') + key return crypto.createHash('md5').update(query, 'utf8').digest('hex') } class CreateOrder extends API { constructor() { super() this.setPath('/Order/CreateOrder') this.setMethod('POST') // 启动订单支付 MQ 消费者(只会启动一次) startOrderPaymentWorker(this.logger) } async onRequest(req, res) { const { uuid, session, goods_id, pay_type } = req.body if ([uuid, session, goods_id, pay_type].some(v => v === '' || v === null || v === undefined)) { return res.json({ ...BaseStdResponse.MISSING_PARAMETER }) } const sessionValid = await AccessControl.checkSession(uuid, session) if (!sessionValid) { return res.status(401).json({ ...BaseStdResponse.ACCESS_DENIED }) } try { const goodsSql = 'SELECT name, price, num, state FROM goods WHERE id = ?' const goodsRows = await db.query(goodsSql, [goods_id]) if (!goodsRows || goodsRows.length !== 1) { return res.json({ ...BaseStdResponse.ERR, msg: '商品不存在' }) } const goods = goodsRows[0] if (goods.num < 1 || goods.state !== 1) { return res.json({ ...BaseStdResponse.ERR, msg: '商品已下架或库存不足' }) } const createTime = Date.now() const orderId = generateOrderId() const insertSql = ` INSERT INTO orders (orderId, create_user, create_time, goods_id, price, pay_type) VALUES (?, ?, ?, ?, ?, ?) ` const result = await db.query(insertSql, [ orderId, uuid, createTime, goods_id, goods.price, pay_type ]) const updateSql = 'UPDATE goods SET num = num - 1 WHERE id = ?' await db.query(updateSql, [goods_id]) if (result && result.affectedRows > 0) { const paymentConfig = config.pay || {} if (!paymentConfig.pid || !paymentConfig.url || !paymentConfig.key || !paymentConfig.return_url) { return res.json({ ...BaseStdResponse.ERR, msg: '支付配置错误,请联系管理员' }) } const deviceType = req.headers['device-type'] ?? '浏览器' let return_url if(deviceType === '哪吒乐跑 Uniapp Client') return_url = paymentConfig.uni_return_url + orderId else return_url = paymentConfig.return_url + orderId const payParams = { pid: paymentConfig.pid, type: pay_type, out_trade_no: orderId, notify_url: `${config.url}/Order/CallBack`, return_url, name: goods.name, money: goods.price } const sign = generatePaymentSign(payParams, paymentConfig.key) payParams.sign = sign payParams.sign_type = 'MD5' await Redis.set(`payData:${orderId}`, JSON.stringify(payParams), { EX: 300 }) // 下单成功后推送到订单支付检查队列,由当前模块的消费者进行轮询及超时取消 try { const ch = await mq.getChannel('order_payment') await ch.assertQueue('order_payment_check', { durable: true }) const msg = { orderId, enqueueTime: Date.now() } await mq.sendToQueueSafe( 'order_payment', 'order_payment_check', Buffer.from(JSON.stringify(msg)), { persistent: true, contentType: 'application/json' } ) } catch (error) { this.logger.error(`推送订单支付检查消息到 MQ 失败,订单号:${orderId},错误:${error.stack || error}`) } res.json({ ...BaseStdResponse.OK, id: orderId, pay: { payUrl: `${paymentConfig.url}/submit.php`, payData: payParams } }) } else { return res.json({ ...BaseStdResponse.ERR, msg: '创建订单失败' }) } } catch (err) { this.logger.error(`创建订单失败!${err.stack}`) return res.json({ ...BaseStdResponse.ERR, msg: "创建订单异常,请联系管理员" }) } } } module.exports.CreateOrder = CreateOrder