ic-ctbu-backend/lib/Lepao/Worker.js

const path = require('path')
const OSS = require('ali-oss')
const mq = require('../../plugin/mq')
const { mq: mqName } = require('../../plugin/mq/mqPrefix')
const { assertRunforgeTaskIngress, TASK_QUEUE } = require('../../plugin/mq/runforgeTaskMq')
const db = require('../../plugin/DataBase/db')
const Redis = require('../../plugin/DataBase/Redis')
const EmailTemplate = require('../../plugin/Email/emailTemplate')
const { URLSearchParams } = require('url')
const {
    getPathData,
    selectCheckpoints,
    generateCadence
} = require('../../plugin/Lepao/Path')
const { v4: uuidv4 } = require('uuid')
const {
    dataEncrypt,
    dataDecrypt,
    dataSign
} = require('../../plugin/Lepao/Crypto')
const generateGyrFromPath = require('../../plugin/Lepao/generateGyrFromPath')
const { syncAccountInfo } = require('./syncAccountInfo')
const { postLepaoSchool } = require('./lepaoSchoolHttp')
const QgProxyManager = require('./QgProxyManager')
const { insertLedgerRecord } = require('./CountLedger')

const Logger = require('../Logger')

class Worker {
    constructor() {
        this.logger = new Logger(
            path.join(__dirname, '../logs/LepaoWorker.log'),
            'INFO'
        )

        this.handlers = {}
        this.running = false

        this.baseUrl = 'https://lepao.ctbu.edu.cn/v3/api.php'

        this.taskQueue = TASK_QUEUE
        this.resultQueue = mqName('runforge_task_result_queue')
        this.deadQueue = mqName('runforge_task_dead_queue')
        this.noticeQueue = mqName('runforge_message_queue')

        this.channelName = 'lepao_worker'

        this.maxRetry = 3
        /** 单次学校接口 HTTP 超时（用于 this.request 内部） */
        this.httpTimeoutMs = 30000
        /** 单个 MQ 任务总超时（覆盖 startRun 多阶段 + 重试） */
        this.taskTimeoutMs = 180000
        this.maxQueueLength = 2000

        this.defaultUserAgent = 'Mozilla/5.0 (Linux; Android 16; 2211133C Build/BP2A.250605.031.A3; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/138.0.7204.180 Mobile Safari/537.36 XWEB/1380347 MMWEBSDK/20250202 MMWEBID/1020 wxwork/5.0.6.66174 MicroMessenger/8.0.28.48(0x28001c30) MiniProgramEnv/android Luggage/3.0.2.95ef3f83 NetType/WIFI Language/zh_CN ABI/arm64'

        // HTTP 出站（charles/fiddler / 青果）由 lepaoSchoolHttp 统一管理：LEPAO_DEBUG_PROXY=1 时强制走本地调试代理。
    }

    /* ================= 工具 ================= */

    formatYmdLocal(d = new Date()) {
        const yyyy = d.getFullYear()
        const mm = String(d.getMonth() + 1).padStart(2, '0')
        const dd = String(d.getDate()).padStart(2, '0')
        return `${yyyy}-${mm}-${dd}`
    }

    simpleHash(str) {
        const s = String(str || '')
        let h = 0
        for (let i = 0; i < s.length; i++) {
            h = ((h << 5) - h + s.charCodeAt(i)) | 0
        }
        return (h >>> 0).toString(16)
    }

    isCommonLepaoErrorMessage(msg) {
        const m = String(msg || '')
        return (
            m.includes('当天已乐跑成功') ||
            m.includes('当前不在有效乐跑时间范围内') ||
            // 登录失效/过期（多种文案变体）
            m.includes('乐跑账号登录已过期') ||
            m.includes('登录已过期') ||
            m.includes('登录信息失效') ||
            m.includes('登录失效') ||
            m.includes('登录状态异常') ||
            m.includes('token已过期') ||
            m.includes('token 过期') ||
            m.includes('Token已过期') ||
            m.includes('Token 过期') ||
            // 兜底：出现“重新登录”通常也是登录态问题
            m.includes('请重新登录')
        )
    }

    async notifyAdminsUncommonLepaoError(payload) {
        try {
            const { account, traceId, taskType, reason, code, name } = payload || {}
            if (!account || !reason) return
            if (this.isCommonLepaoErrorMessage(reason)) return

            const day = this.formatYmdLocal()
            const msgHash = this.simpleHash(reason)
            const rateKey = `lepao:adminWarn:${account}:${day}:${msgHash}`
            const existed = await Redis.get(rateKey)
            if (existed) return
            await Redis.set(rateKey, '1', { EX: 86400 })

            const adminSql = `
                SELECT email
                FROM users
                WHERE email IS NOT NULL
                  AND email <> ''
                  AND (JSON_CONTAINS(permission, '"admin"') OR JSON_CONTAINS(permission, '"service"'))
            `
            const adminRows = await db.query(adminSql)
            if (!adminRows || adminRows.length === 0) {
                this.logger.warn(`[lepaoAdminWarn][${traceId}] 未找到可通知的管理员邮箱`)
                return
            }

            const emails = [...new Set(adminRows.map(r => r.email).filter(Boolean))]
            for (const email of emails) {
                await EmailTemplate.lepaoAdminWarning(email, {
                    server: (() => {
                        try {
                            // config.json 可能在其他模块缓存读取，这里动态 require 一次避免循环依赖
                            // eslint-disable-next-line global-require
                            const cfg = require('../../config.json')
                            return cfg?.server || ''
                        } catch (_) {
                            return ''
                        }
                    })(),
                    account,
                    name: name || '',
                    taskType: taskType || 'lepao.startRun',
                    traceId: traceId || '',
                    reason,
                    code: code || ''
                })
            }
        } catch (e) {
            this.logger.error(`[lepaoAdminWarn] 告警邮件发送失败：${e.message || e}`)
        }
    }

    api(path) {
        return this.baseUrl + path
    }

    traceId() {
        return Date.now() + '_' + Math.random().toString(36).slice(2, 8)
    }

    sleep(ms) {
        return new Promise(r => setTimeout(r, ms))
    }

    buildOssBaseClientConfig(sts) {
        return {
            bucket: sts.bucket,
            region: sts.region || 'oss-cn-hangzhou',
            accessKeyId: sts.AccessKeyId,
            accessKeySecret: sts.AccessKeySecret,
            stsToken: sts.SecurityToken,
            secure: true
        }
    }

    buildProxyUrlFromFragment(fragment) {
        if (!fragment || fragment.proxy === false || !fragment.proxy) return null
        const { host, port, auth } = fragment.proxy
        let userPart = ''
        if (auth && String(auth.username || '').length > 0) {
            const u = encodeURIComponent(auth.username)
            const p = encodeURIComponent(auth.password != null ? String(auth.password) : '')
            userPart = `${u}:${p}@`
        }
        return `http://${userPart}${host}:${port}`
    }

    async putOssWithFallback(sts, ossPath, content) {
        const directClient = new OSS(this.buildOssBaseClientConfig(sts))
        await directClient.put(ossPath, content)
    }

    isRunSuccess(bindResponse) {
        const payload = bindResponse?.data
        if (!bindResponse || bindResponse.status !== 1 || !payload) {
            return {
                ok: false,
                reason: bindResponse?.info || '系统繁忙，请联系客服或稍后再试'
            }
        }

        const failedReason = payload.record_failed_reason || ''
        if (failedReason === '' || failedReason === '自动确认有效') {
            return { ok: true, payload }
        }

        return {
            ok: false,
            reason: failedReason,
            payload
        }
    }

    maskClientReason(reason) {
        const s = String(reason || '')
        if (s.includes('非法请求')) {
            return '系统繁忙，请稍后再试'
        }
        return s
    }

    extractApiErrorMessage(name, result) {
        if (!result) {
            this.logger.error(`${name} 接口无响应数据: ${this.safeStringify(result)}`)
            return `系统繁忙，请联系客服或稍后再试`
        }

        const candidates = [
            result.info,
            result.msg,
            result.message,
            result?.data?.info,
            result?.data?.msg,
            result?.data?.message,
            result?.data?.record_failed_reason
        ]

        const reason = candidates.find(v => typeof v === 'string' && v.trim() !== '')
        if (reason) {
            return this.maskClientReason(reason)
        }

        if (result.code !== undefined || result.status !== undefined) {
            this.logger.error(`${name} 接口返回异常: ${this.safeStringify(result)}`)
            return `系统繁忙，请联系客服或稍后再试`
        }
        return `系统繁忙，请联系客服或稍后再试`
    }

    async markLoginExpired(account) {
        if (!account) return
        try {
            const sql = 'UPDATE lepao_account SET state = 0 WHERE student_num = ?'
            await db.query(sql, [account])
            this.logger.warn(`${account} 登录状态已失效，已自动更新为未登录`)
        } catch (error) {
            this.logger.error(`更新账号登录状态失败：${error.stack || error}`)
        }
    }

    async writeSuccessRedis(account) {
        if (!account) return
        try {
            const now = new Date()
            const tomorrow = new Date().setHours(24, 0, 0, 0)
            const exp = Math.floor((tomorrow - now) / 1000)
            await Redis.set(`lepaoSuccess:${account}`, account, { EX: exp })
        } catch (error) {
            this.logger.error(`写入乐跑成功缓存失败: ${error.stack || error}`)
        }
    }

    async addLepaoRecord(uuid, account, result, pathId, pointData, runMode = 'auto') {
        if (!uuid || !account || !result || !pathId) return
        try {
            const time = Date.now()
            const publicId = uuidv4()
            const safeRunMode = runMode === 'manual' ? 'manual' : 'auto'
            const sql = 'INSERT INTO lepao_record (public_id, uuid, time, lepao_account, result, path_id, point_data, state, run_mode) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)'
            await db.query(sql, [publicId, uuid, time, account, result, pathId, JSON.stringify(pointData || []), 1, safeRunMode])
        } catch (error) {
            this.logger.error(`写入乐跑记录失败: ${error.stack || error}`)
        }
    }

    async syncRunCount(req, ctx) {
        try {
            const sid = req?.student_id || req?.account
            if (
                req?.uid == null ||
                req?.token == null ||
                String(req.token).trim() === '' ||
                req?.school_id == null ||
                !sid
            ) {
                return { ok: false, reason: '缺少同步乐跑次数所需凭证' }
            }

            const recordData = await this.handlers['lepao.getRecord'](req, ctx)
            const data = recordData?.data
            if (!data) {
                return { ok: false, reason: 'getRecord 无有效 data' }
            }

            const term_num = Number(data.term_num ?? 30)
            const total_num = Number(data.total_num ?? 0)

            const sql = 'UPDATE lepao_account SET term_num = ?, total_num = ? WHERE student_num = ?'
            const rows = await db.query(sql, [term_num, total_num, req.account])
            if (!rows || rows.affectedRows !== 1) {
                this.logger.warn(`${req.account}更新乐跑次数失败`)
                return { ok: false, reason: '数据库更新 lepao_account 失败', term_num, total_num }
            }
            this.logger.info(`${req.account}更新乐跑次数成功 term_num=${term_num}, total_num=${total_num}`)
            return { ok: true, term_num, total_num }
        } catch (error) {
            this.logger.warn(`${req?.account || 'unknown'}同步乐跑次数失败: ${error.message || error}`)
            return { ok: false, reason: error.message || String(error) }
        }
    }

    lepaoTimestamp() {
        return Number((Date.now() / 1000).toFixed(3))
    }

    async enqueueTask(channel, type, data, options = {}) {
        const payload = {
            id: options.id || this.traceId(),
            type,
            data,
            retry: options.retry ?? 0
        }

        await channel.sendToQueue(
            this.taskQueue,
            Buffer.from(JSON.stringify(payload)),
            { persistent: true, contentType: 'application/json' }
        )

        return payload.id
    }

    async withTimeout(promise, name, timeoutMs, options = {}) {
        const ms = Number.isFinite(Number(timeoutMs)) ? Number(timeoutMs) : this.taskTimeoutMs
        const { code = 'TIMEOUT', retryable = false } = options || {}
        return Promise.race([
            promise,
            new Promise((_, reject) =>
                setTimeout(() => {
                    const err = new Error(`${name} 超时`)
                    err.code = code
                    err.retryable = !!retryable
                    reject(err)
                }, ms)
            )
        ])
    }

    async retry(fn, name) {
        let lastErr

        for (let i = 0; i < this.maxRetry; i++) {
            try {
                return await fn()
            } catch (err) {
                lastErr = err
                if (!this.isRetryableTaskError(err)) {
                    throw err
                }
                this.logger.warn(`[RETRY] ${name} 第${i + 1}次失败`)

                await this.sleep(1000 * (i + 1)) // 指数退避
            }
        }

        throw lastErr
    }

    isNetworkError(err) {
        if (!err) return false
        if (err.code && ['ECONNRESET', 'ECONNABORTED', 'ETIMEDOUT', 'ENOTFOUND', 'EAI_AGAIN'].includes(err.code)) {
            return true
        }
        if (err.isAxiosError && !err.response) return true
        const msg = (err.message || '').toLowerCase()
        return msg.includes('timeout') || msg.includes('network')
    }

    isRetryableTaskError(err) {
        if (!err) return false
        if (err.retryable === true) return true
        if (this.isNetworkError(err)) return true
        return ['PATH_SELECT_FAILED', 'CHECKPOINT_FETCH_FAILED', 'CHECKPOINT_INSUFFICIENT'].includes(err.code)
    }

    safeStringify(obj) {
        const seen = new WeakSet();
        return JSON.stringify(obj, (key, value) => {
            if (typeof value === 'object' && value !== null) {
                if (seen.has(value)) return '[Circular]';
                seen.add(value);
            }
            return value;
        })
    }

    log(traceId, type, msg, data) {
        this.logger.info(`[${traceId}] [${type}] ${msg} ${data ? this.safeStringify(data) : ''}`)
    }

    logErr(traceId, msg, err) {
        this.logger.error(`[${traceId}] ${msg} ${err.stack || err}`)
    }

    async request(traceId, name, url, raw, headers = {}, ctx = null) {
        return this.retry(async () => {
            this.log(traceId, 'REQ', name, raw)

            const mergedHeaders = {
                'Content-Type': 'application/x-www-form-urlencoded',
                'Accept': '*/*',
                'Accept-Language': 'zh-CN,zh-Hans;q=0.9',
                'Accept-Encoding': 'gzip, deflate, br',
                'Referer': 'https://servicewechat.com/wxf94c4ddb63d87ede/32/page-frame.html',
                ...headers
            }
            if (!mergedHeaders['User-Agent']) {
                mergedHeaders['User-Agent'] = this.defaultUserAgent
            }

            const form = new URLSearchParams()
            form.append('ostype', '5')
            form.append('data', dataEncrypt(JSON.stringify(raw)))

            const res = await this.withTimeout(
                postLepaoSchool(url, form, {
                    headers: mergedHeaders,
                    timeout: this.httpTimeoutMs,
                    logger: this.logger,
                    outboundMode: ctx?.outboundMode || 'auto',
                    mqTaskId: ctx?.taskId
                }),
                name,
                this.httpTimeoutMs,
                { code: 'HTTP_TIMEOUT', retryable: true }
            )

            let result = res.data

            if (result?.data && result?.is_encrypt === 1) {
                result.data = JSON.parse(dataDecrypt(result.data))
            }

            this.log(traceId, 'RES', name, result)

            // 除 bindData 外，其余调用若接口已明确返回失败，直接抛出该失败原因
            // bindData 需要保留完整响应由 isRunSuccess 统一判定。
            if (name !== 'bindData') {
                const hasCode = result && Object.prototype.hasOwnProperty.call(result, 'code')
                const hasStatus = result && Object.prototype.hasOwnProperty.call(result, 'status')
                const failedByCode = hasCode && Number(result.code) !== 1 && Number(result.code) !== 200
                const failedByStatus = hasStatus && Number(result.status) !== 1
                if (failedByCode || failedByStatus) {
                    const message = this.extractApiErrorMessage(name, result)
                    const err = new Error(message)
                    // 学习 Lepao.js：若明确提示重新登录，自动标记账号失效
                    if (message.includes('重新登录')) {
                        await this.markLoginExpired(raw?.student_num)
                    }
                    // 接口已返回业务错误，禁止重试
                    err.retryable = false
                    throw err
                }
            }

            return result
        }, name)
    }

    /**
     * 累计完成次数 >= 跑友目标（且目标>0）时：关闭 auto_run，并发送乐跑目标完成邮件 / Bot 通知
     */
    async handleLepaoTargetComplete(account, user, totalNum, traceId) {
        const target = Number(user?.target_count) || 0
        const total = Number(totalNum) || 0
        if (target <= 0 || total < target) return

        try {
            const up = await db.query(
                'UPDATE lepao_account SET auto_run = 0 WHERE student_num = ? AND auto_run = 1',
                [account]
            )
            if (up?.affectedRows !== 1) {
                return
            }
            this.logger.info(
                `${account} 已达目标次数（${total}/${target}），关闭自动乐跑`
            )
        } catch (e) {
            this.logger.error(`关闭自动乐跑失败 ${account}: ${e.message || e}`)
            return
        }

        const noticeType = user.notice_type || 'none'
        const overPayload = {
            type: 'lepao_over',
            umo: user.bot_umo,
            name: user.name,
            account,
            total_num: total,
            target_count: target,
            traceId
        }

        if (noticeType === 'email' && user.email) {
            try {
                await EmailTemplate.lepaoOver(user.email, {
                    name: user.name,
                    account
                })
            } catch (e) {
                this.logger.error(`lepaoOver 邮件发送失败: ${e.message || e}`)
            }
        }

        if (noticeType === 'bot' && user.bot_umo) {
            try {
                const ch = await mq.getChannel(this.noticeQueue)
                await ch.assertQueue(this.noticeQueue, { durable: true })
                ch.sendToQueue(
                    this.noticeQueue,
                    Buffer.from(JSON.stringify(overPayload)),
                    {
                        persistent: true,
                        contentType: 'application/json'
                    }
                )
            } catch (e) {
                this.logger.error(`lepao_over Bot 通知失败: ${e.message || e}`)
            }
        }
    }

    register(type, handler) {
        this.handlers[type] = handler
        this.logger.info(`注册任务: ${type}`)
    }

    /* ================= 业务 ================= */

    initHandlers() {
        this.register('lepao.syncAccountInfo', async (req) => {
            const studentNum = req?.student_num
            if (!studentNum) {
                throw new Error('同步乐跑账号失败：缺少 student_num')
            }
            const syncResult = await syncAccountInfo({
                studentNum,
                logger: this.logger
            })
            if (!syncResult.ok) {
                const err = new Error(syncResult.msg || '同步乐跑账号失败')
                err.retryable = false
                throw err
            }
            return syncResult.data
        })

        /* ---------------- 开始乐跑 ---------------- */
        this.register('lepao.startRun', async (req, ctx) => {
            const traceId = ctx.traceId
            const maxPathRetry = 20  // 自动获取路径失败最大重试次数
            let pathRetry = 0
            let pointData = null
            let pathData = null
            let newPathData = null
            let ossPath = null
            let ossSts = null
            let userData = null
            let pathId = null
            let runZoneId = 0
            let bindRes = null

            try {
                // 检查redis是否存在当天乐跑成功记录
                const isSuccess = await Redis.get(`lepaoSuccess:${req.account}`)
                if (isSuccess)
                    throw new Error('该账号当天已乐跑成功！请勿重复乐跑')

                userData = await this.handlers['lepao.getUserData'](req, ctx)


                // 立刻合并账号凭证，保证后续任意 throw 时 finally 里 syncRunCount 不会用空 token 调 getRecord
                req = {
                    ...req,
                    ...userData,
                    student_id: req.account
                }

                // 进入乐跑进程后写入进行中缓存
                const progressKey = `lepaoProgress:${req.account}`
                const inProgress = await Redis.get(progressKey)
                if (inProgress) {
                    throw new Error('该账号已进入乐跑任务队列，请等待乐跑完成后再进行乐跑操作')
                }
                await Redis.set(progressKey, req.account, { EX: 1800 })

                ossSts = await this.handlers['lepao.getOssSts'](req, ctx)
                if (!ossSts?.bucket || !ossSts?.AccessKeyId || !ossSts?.AccessKeySecret || !ossSts?.SecurityToken) {
                    throw new Error('获取 OSS 凭证失败，请联系客服或稍后再试')
                }

                req = {
                    ...req,
                    ossSts
                }

                // 晚上10点后提前
                let run_end_time = Math.floor(Date.now() / 1000) - 300 // 提前5分钟
                let hour = new Date().getHours()

                if (hour < 7)
                    throw new Error('当前不在有效乐跑时间范围内。RunForge支持乐跑时间段为7:00~24:00')

                if (hour >= 22) {
                    this.logger.info(`${req.account}当前时间为${hour}点，调整run_end_time提前5小时`)
                    run_end_time -= 18000
                }

                req = {
                    ...req,
                    run_end_time
                }

                // 1.5️⃣ 乐跑开始前扣减次数（失败会返还，且有幂等保护）
                await this.handlers['lepao.consumeCount']({
                    account: req.account,
                    uuid: userData?.create_user
                }, ctx)

                while (pathRetry < maxPathRetry) {
                    try {
                        // 2️⃣ 获取路径（仅路径选择失败时重试）
                        const pathRes = await this.handlers['lepao.getPath'](req, ctx)
                        pathData = pathRes.pathData
                        pathId = pathData?.id || null
                        newPathData = getPathData(pathData.data, req.run_end_time, pathData.time)

                        // 3️⃣ 切换跑区
                        const zoneRes = await this.handlers['lepao.setZone']({ ...req, pathData }, ctx)
                        runZoneId = zoneRes?.run_zone_id || 0

                        // 4️⃣ 上传 OSS 文件、生成打卡点
                        const uploadRes = await this.handlers['lepao.uploadOssFile']({ ...req, pathData, newPathData }, ctx)
                        ossPath = uploadRes.oss_path
                        pointData = uploadRes.point_data

                        if (!pointData) {
                            pathRetry++
                            this.logger.warn(`[${traceId}] 打卡点不满足要求，重新获取路径 第${pathRetry}次`)
                            continue
                        }

                        // 打卡点符合要求，跳出循环
                        break
                    } catch (err) {
                        if (!this.isRetryableTaskError(err)) {
                            throw err
                        }
                        this.logger.warn(`[${traceId}] 可重试错误，重新获取路径 第${pathRetry + 1}次，原因：${err.message}`)
                        pathRetry++
                        await this.sleep(1000 * pathRetry)
                    }
                }

                if (!pointData) {
                    throw new Error('打卡点获取失败，乐跑任务终止')
                }

                // 5️⃣ 提交跑步数据
                bindRes = await this.handlers['lepao.bindData']({
                    ...req,
                    pathData,
                    run_zone_id: runZoneId,
                    record_file: ossPath,
                    point_data: pointData
                }, ctx)

                // 绑定接口有返回即入库
                if (bindRes && bindRes.data) {
                    await this.addLepaoRecord(userData?.create_user, req.account, bindRes.data, pathId, pointData, req.runMode)
                }

                const runResult = this.isRunSuccess(bindRes)
                if (runResult.ok || runResult.reason === '当天关联成绩次数已达到上限') {
                    await this.writeSuccessRedis(req.account)
                }
                if (!runResult.ok) {
                    throw new Error(this.maskClientReason(runResult.reason))
                }
                if (bindRes && bindRes.data && bindRes.data.record_id) {
                    const gyrRes = await this.handlers['lepao.uploadGyrOssFile']({ ...req, newPathData, record_id: bindRes.data.record_id }, ctx)
                    if (gyrRes?.status === 1) {
                        this.logger.info(`${req.account}上传加速度数据成功！`)
                    } else {
                        this.logger.error(`${req.account}上传加速度数据失败！原因：${gyrRes.info || '未知错误'}`)
                    }
                }

                // 同步乐跑次数（通知里要带 total_num / term_num，与 getRecord 一致）
                const syncResult = await this.syncRunCount(req, ctx)
                if (!syncResult?.ok) {
                    this.logger.error(
                        `[${traceId}] 同步乐跑次数失败：${syncResult?.reason || 'unknown'}`
                    )
                }

                // 6️⃣ 发送通知（把同步后的学期次数、累计完成次数传给 Bot / 邮件）
                if (ctx.channel) {
                    await this.enqueueTask(
                        ctx.channel,
                        'lepao.sendNotice',
                        {
                            account: req.account,
                            success: true,
                            data: runResult.payload,
                            traceId,
                            total_num: syncResult?.ok ? syncResult.total_num : undefined
                        },
                        { id: `${traceId}:notice:success` }
                    )
                }

                return { traceId, ossPath, pointData, bindRes }

            } catch (err) {
                this.logger.error(`[${traceId}] 乐跑流程失败:`, err)

                const retryNow = Number(ctx?.retry || 0)
                const maxRetry = Number(ctx?.maxRetry ?? this.maxRetry)
                const canRetry = this.isRetryableTaskError(err) && retryNow < maxRetry
                const isFinalAttemptFail = !canRetry

                // 仅在所有重试均失败后才发异常告警/失败通知，避免每次重试都打扰
                if (isFinalAttemptFail) {
                    Promise.resolve().then(async () => {
                        try {
                            await this.notifyAdminsUncommonLepaoError({
                                account: req?.account,
                                name: userData?.name,
                                traceId,
                                taskType: 'lepao.startRun',
                                reason: err?.message || '未知错误',
                                code: err?.code
                            })
                        } catch (_) { }
                    })
                }

                // 若已扣减次数，则失败时返还（幂等）
                try {
                    await this.handlers['lepao.refundCount']({
                        account: req.account,
                        uuid: userData?.create_user
                    }, ctx)
                } catch (e) {
                    this.logger.error(`[${traceId}] 返还乐跑次数失败：${e.stack || e}`)
                }

                if (ctx.channel && isFinalAttemptFail) {
                    await this.enqueueTask(ctx.channel, 'lepao.sendNotice', {
                        account: req.account,
                        success: false,
                        reason:
                            err?.code === 'TASK_TIMEOUT'
                                ? '系统繁忙，请稍后再试'
                                : this.maskClientReason(err.message || '未知错误'),
                        traceId
                    }, { id: `${traceId}:notice:fail` })
                }

                throw err
            } finally {
                await Redis.del(`lepaoProgress:${req.account}`)
            }
        })

        /* ---------------- 发送通知（独立 MQ 任务） ---------------- */
        this.register('lepao.sendNotice', async (req, ctx) => {
            const { account, success, data, reason, traceId, total_num: totalNumArg } = req || {}

            if (!account) {
                throw new Error('发送通知失败：缺少 account')
            }

            const emailSql = `
                SELECT 
                    a.name, 
                    a.email, 
                    a.target_count,
                    a.notice_type,
                    e.bot_umo
                FROM 
                    lepao_account a
                LEFT JOIN
                    lepao_extra e
                ON 
                    a.student_num = e.student_num
                WHERE
                    a.student_num = ?
            `
            const rows = await db.query(emailSql, [account])
            if (!rows || rows.length === 0) {
                throw new Error('发送通知失败：未找到用户通知配置')
            }

            const user = rows[0]
            const noticeType = user.notice_type || 'none'

            let totalForNotice = totalNumArg
            if (
                success &&
                (totalForNotice === undefined || totalForNotice === null)
            ) {
                const accRows = await db.query(
                    'SELECT total_num FROM lepao_account WHERE student_num = ?',
                    [account]
                )
                if (accRows && accRows[0]) {
                    totalForNotice = accRows[0].total_num
                }
            }
            if (totalForNotice === undefined || totalForNotice === null) {
                totalForNotice = 0
            }
            totalForNotice = Number(totalForNotice)
            const targetCount = Number(user.target_count) || 0

            const payload = success ? {
                ...(data && typeof data === 'object' ? data : {}),
                type: 'lepao_success',
                umo: user.bot_umo,
                total_num: totalForNotice,
                target_count: targetCount,
                name: user.name,
                account,
                traceId
            } : {
                type: 'lepao_fail',
                umo: user.bot_umo,
                name: user.name,
                account,
                reason,
                traceId
            }

            const afterSuccessNotify = async () => {
                if (success) {
                    await this.handleLepaoTargetComplete(account, user, totalForNotice, traceId)
                }
            }

            if (noticeType === 'bot' && user.bot_umo) {
                const ch = await mq.getChannel(this.noticeQueue)
                await ch.assertQueue(this.noticeQueue, { durable: true })
                ch.sendToQueue(
                    this.noticeQueue,
                    Buffer.from(JSON.stringify(payload)),
                    {
                        persistent: true,
                        contentType: 'application/json'
                    }
                )
                await afterSuccessNotify()
                return { delivered: true, via: 'bot' }
            }

            if (noticeType === 'email' && user.email) {
                if (success) {
                    await EmailTemplate.lepaoSuccess(user.email, payload)
                    await afterSuccessNotify()
                    return { delivered: true, via: 'email' }
                }

                await EmailTemplate.lepaoFail(user.email, {
                    name: user.name,
                    account,
                    reason: reason || '系统繁忙，请联系客服或稍后再试',
                    traceId
                })
                return { delivered: true, via: 'email' }
            }

            await afterSuccessNotify()
            return { delivered: false, via: 'none' }
        })

        /* ---------------- 扣减次数 ---------------- */
        this.register('lepao.consumeCount', async (req, ctx) => {
            const account = req?.account
            const uuid = req?.uuid
            if (!uuid) {
                throw new Error('扣减乐跑次数失败：缺少 uuid')
            }

            // 幂等：同一 taskId 只扣一次
            const consumeKey = `lepao:consume:${ctx?.taskId || ctx?.traceId || account || uuid}`
            const existed = await Redis.get(consumeKey)
            if (existed) {
                return true
            }

            this.logger.info(`${account || uuid}开始扣减乐跑次数`)
            const conn = await db.connect()
            try {
                await conn.beginTransaction()
                const [userRows] = await conn.execute(
                    'SELECT lepao_count FROM users WHERE uuid = ? FOR UPDATE',
                    [uuid]
                )
                if (!userRows || userRows.length !== 1) {
                    await conn.rollback()
                    throw new Error('扣减乐跑次数失败：用户不存在')
                }

                const beforeCount = Number(userRows[0].lepao_count || 0)
                if (beforeCount < 1) {
                    await conn.rollback()
                    throw new Error('用户乐跑次数不足，请购买乐跑次数后重试！')
                }

                const [r] = await conn.execute(
                    'UPDATE users SET lepao_count = lepao_count - 1 WHERE uuid = ?',
                    [uuid]
                )
                if (!r || r.affectedRows !== 1) {
                    await conn.rollback()
                    throw new Error('扣减乐跑次数失败：数据库更新失败')
                }

                await insertLedgerRecord({
                    executor: conn,
                    userUuid: uuid,
                    delta: -1,
                    balanceBefore: beforeCount,
                    balanceAfter: beforeCount - 1,
                    bizType: 'run_consume',
                    bizId: consumeKey,
                    remark: `${account || uuid}乐跑扣除`
                })

                await conn.commit()
            } catch (error) {
                try { await conn.rollback() } catch (_) { }
                throw error
            } finally {
                if (conn?.connection && typeof conn.connection.release === 'function' && typeof conn?.release === 'function') {
                    conn.release()
                }
            }
            this.logger.info(`${account || uuid}扣减乐跑次数完成`)

            await Redis.set(consumeKey, '1', { EX: 3600 })
            return true
        })

        /* ---------------- 返还次数（失败时执行） ---------------- */
        this.register('lepao.refundCount', async (req, ctx) => {
            const account = req?.account
            const uuid = req?.uuid
            if (!uuid) {
                return true
            }

            const baseKey = `${ctx?.taskId || ctx?.traceId || account || uuid}`
            const consumeKey = `lepao:consume:${baseKey}`
            const refundKey = `lepao:refund:${baseKey}`

            const consumed = await Redis.get(consumeKey)
            if (!consumed) {
                return true
            }
            const refunded = await Redis.get(refundKey)
            if (refunded) {
                return true
            }

            this.logger.info(`${account || uuid}开始返还乐跑次数`)
            const conn = await db.connect()
            try {
                await conn.beginTransaction()
                const [userRows] = await conn.execute(
                    'SELECT lepao_count FROM users WHERE uuid = ? FOR UPDATE',
                    [uuid]
                )
                if (!userRows || userRows.length !== 1) {
                    await conn.rollback()
                    throw new Error('返还乐跑次数失败：用户不存在')
                }

                const beforeCount = Number(userRows[0].lepao_count || 0)
                const [r] = await conn.execute(
                    'UPDATE users SET lepao_count = lepao_count + 1 WHERE uuid  = ?',
                    [uuid]
                )
                if (!r || r.affectedRows !== 1) {
                    await conn.rollback()
                    throw new Error('返还乐跑次数失败：数据库更新失败')
                }

                await insertLedgerRecord({
                    executor: conn,
                    userUuid: uuid,
                    delta: 1,
                    balanceBefore: beforeCount,
                    balanceAfter: beforeCount + 1,
                    bizType: 'run_refund',
                    bizId: refundKey,
                    remark: `${account || uuid}乐跑失败返还`
                })

                await conn.commit()
            } catch (error) {
                try { await conn.rollback() } catch (_) { }
                throw error
            } finally {
                if (conn?.connection && typeof conn.connection.release === 'function' && typeof conn?.release === 'function') {
                    conn.release()
                }
            }
            this.logger.info(`${account || uuid}返还乐跑次数完成`)
            await Redis.set(refundKey, '1', { EX: 3600 })
            return true
        })

        this.register('lepao.getUserData', async (req, ctx) => {
            const account = req.account
            this.logger.info(`${account}开始获取用户数据`)
            const accountSql = `
                SELECT
                    u.uuid, 
                    u.lepao_count,
                    l.create_user,
                    l.name, 
                    l.student_num,
                    l.area, 
                    l.sex, 
                    l.state,  
                    l.token, 
                    l.uid, 
                    l.school_id, 
                    l.userAgent,
                    l.deviceModel,
                    l.notice_type, 
                    l.email,
                    e.bot_account
                FROM 
                    lepao_account l
                LEFT JOIN
                    users u
                ON
                    l.create_user = u.uuid
                LEFT JOIN
                    lepao_extra e
                ON 
                    l.student_num = e.student_num
                WHERE 
                    l.student_num = ?
            `
            const rows = await db.query(accountSql, [account])
            if (!rows || rows.length === 0) {
                this.logger.error(`${account}无法获取账号数据`)
                throw new Error('无法获取账号数据，请联系客服或稍后再试')
            }

            let userData = rows[0]

            if (!userData.create_user || !userData.uuid) {
                this.logger.warn(`${account}账号状态异常`)
                throw new Error('当前账号状态异常，请联系客服')
            }

            if (userData.state !== 1) {
                this.logger.warn(`${account}登录状态异常 state=${userData.state}`)
                throw new Error('乐跑账号登录已过期，请尝试使用登录器重新登录')
            }

            if (userData.lepao_count < 1) {
                this.logger.warn(`${account}乐跑次数不足`)
                throw new Error('用户乐跑次数不足，请购买乐跑次数后重试！')
            }

            if (!userData.userAgent)
                userData.userAgent = 'Mozilla/5.0 (Linux; Android 16; 2211133C Build/BP2A.250605.031.A3; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/138.0.7204.180 Mobile Safari/537.36 XWEB/1380347 MMWEBSDK/20250202 MMWEBID/1020 wxwork/5.0.6.66174 MicroMessenger/8.0.28.48(0x28001c30) MiniProgramEnv/android Luggage/3.0.2.95ef3f83 NetType/WIFI Language/zh_CN ABI/arm64'

            if (!userData.deviceModel)
                userData.deviceModel = '2211133C'

            return userData
        })

        this.register('lepao.getPath', async (req, ctx) => {
            const account = req.account
            this.logger.info(`${account}开始获取路径`)
            const accountSql = 'SELECT area, sex FROM lepao_account WHERE student_num = ?'
            const rows = await db.query(accountSql, [account])
            if (!rows || rows.length === 0) {
                this.logger.error(`${account}无法获取账号数据`)
                throw new Error('无法获取账号数据')
            }

            const { area, sex } = rows[0]

            let max = 4.00
            let min = 2.00
            if (sex === 2) {
                max = 2.00
                min = 1.60
            }

            this.logger.info(`${account}路径参数: area=${area ?? '随机'}, max_distance=${max}, min_distance=${min}`)

            let pathSql = 'SELECT * FROM path_data WHERE state = 1 AND distance < ? AND distance > ? '
            const pathParams = [max, min]

            if (area) {
                pathSql += ' AND run_zone_name = ?'
                pathParams.push(area)
            }

            pathSql += ' ORDER BY count ASC LIMIT 1'

            const paths = await db.query(pathSql, pathParams)
            if (!paths || paths.length === 0) {
                this.logger.error(`${account}未找到符合条件的路线`)
                const err = new Error('未找到符合条件的路线，请改变路径选择条件')
                err.code = 'PATH_SELECT_FAILED'
                err.retryable = true
                throw err
            }

            const randomPath = paths[0]

            const updateSql = 'UPDATE path_data SET count = count + 1 WHERE id = ?'
            await db.query(updateSql, [randomPath.id])

            this.logger.info(`${account}路径选中id=${randomPath.id}，计数加1成功`)

            return { pathData: randomPath }
        })

        /* ---------------- 获取跑步记录 ---------------- */
        this.register('lepao.getRecord', async (req, ctx) => {
            const now = this.lepaoTimestamp()
            const raw = {
                uid: req.uid,
                token: req.token,
                school_id: req.school_id,
                term_id: 0,
                course_id: 0,
                class_id: 0,
                student_num: req.student_id,
                card_id: req.student_id,
                timestamp: now,
                version: 1,
                nonce: String(Math.floor(Math.random() * 900000 + 100000)),
                ostype: 5
            }
            raw.sign = dataSign(raw)

            return this.request(
                ctx.traceId,
                'getRecord',
                this.api('/Run2/beforeRunV260'),
                raw,
                {
                    'User-Agent': req.userAgent,
                    'charset': 'utf-8',
                    'Referer': 'https://servicewechat.com/wxf94c4ddb63d87ede/32/page-frame.html',
                },
                ctx
            )
        })

        /* ---------------- 切换跑区 ---------------- */
        this.register('lepao.setZone', async (req, ctx) => {
            const runZoneMap = {
                '兰花湖校区跑区': 2,
                '主校区北跑区': 3,
                '主校区南跑区': 5,
                '重庆工商大学茶园校区': 6
            }

            const { pathData } = req

            const runZoneId = runZoneMap[pathData.run_zone_name]
            if (!runZoneId) throw new Error('跑区不存在')

            const raw = {
                uid: req.uid,
                token: req.token,
                school_id: req.school_id,
                term_id: 0,
                course_id: 0,
                class_id: 0,
                student_num: req.student_id,
                card_id: req.student_id,
                timestamp: this.lepaoTimestamp(),
                version: 1,
                nonce: String(Math.floor(Math.random() * 900000 + 100000)),
                ostype: 5,
                run_zone_id: String(runZoneId)
            }
            raw.sign = dataSign(raw)

            await this.request(
                ctx.traceId,
                'setZone',
                this.api('/Run/setRunZone'),
                raw,
                {},
                ctx
            )
            return { run_zone_id: runZoneId }
        })

        /* ---------------- 获取 OSS STS ---------------- */
        this.register('lepao.getOssSts', async (req, ctx) => {
            const raw = {
                uid: req.uid,
                token: req.token,
                school_id: req.school_id,
                term_id: 0,
                course_id: 0,
                class_id: 0,
                student_num: req.student_id,
                card_id: req.student_id,
                timestamp: this.lepaoTimestamp(),
                version: 1,
                nonce: String(Math.floor(Math.random() * 900000 + 100000)),
                ostype: 5
            }
            raw.sign = dataSign(raw)

            const res = await this.request(
                ctx.traceId,
                'getOssSts',
                this.api('/WpIndex/getOssSts'),
                raw,
                {},
                ctx
            )

            return res.data
        })

        /* ---------------- 上传 OSS 文件 ---------------- */
        this.register('lepao.uploadOssFile', async (req, ctx) => {
            const { pathData, newPathData, ossSts: sts } = req

            // 处理跑步路径
            const pathResult = dataEncrypt(JSON.stringify(newPathData))

            // 获取跑步规则参数
            const runRule = await this.handlers['lepao.getRecord'](req, ctx)
            const ruleData = runRule?.data
            if (!ruleData?.run_line_info?.point_list || !ruleData?.time_rule_arr?.length) {
                const err = new Error('获取打卡点规则失败')
                err.code = 'CHECKPOINT_FETCH_FAILED'
                err.retryable = true
                throw err
            }
            const check_points = ruleData.run_line_info.point_list
            let min_log_num = ruleData.time_rule_arr[0]?.min_log_num || 4
            const point_update_distance = parseFloat(ruleData.run_line_info.point_update_distance || 0) * 1000
            const log_max_distance = Number(ruleData.run_line_info.log_max_distance || 0)

            // 生成打卡点
            const point_data = selectCheckpoints(newPathData, check_points, min_log_num, point_update_distance, log_max_distance, req.run_end_time, pathData.time)
            if (!point_data) {
                this.logger.warn(`[RETRY] 打卡点数量不足，重新更换路径`)
                const err = new Error('打卡点数量不足')
                err.code = 'CHECKPOINT_INSUFFICIENT'
                err.retryable = true
                throw err
            }

            const now = new Date()
            const yyyy = now.getFullYear()
            const mm = String(now.getMonth() + 1).padStart(2, '0')
            const dd = String(now.getDate()).padStart(2, '0')
            const formattedToday = `${yyyy}-${mm}-${dd}`
            const boundary = String(Date.now())
            const timestamp = String(Date.now())
            const ossPath = `Public/Upload/file/run_record/${boundary.slice(-3)}/${formattedToday}/${timestamp}-${Math.floor(Math.random() * 150)}.txt`
            await this.putOssWithFallback(sts, ossPath, Buffer.from(pathResult, 'utf-8'))

            return { oss_path: ossPath, point_data: point_data }
        })

        this.register('lepao.uploadGyrOssFile', async (req, ctx) => {
            const { newPathData, ossSts: sts, record_id } = req

            // 生成加速度数据
            const gyrData = generateGyrFromPath(newPathData)
            if (!Array.isArray(gyrData) || gyrData.length === 0) {
                this.logger.error('生成加速度数据失败')
                return { status: 0, info: '生成加速度数据失败' }
            }

            const now = new Date()
            const yyyy = now.getFullYear()
            const mm = String(now.getMonth() + 1).padStart(2, '0')
            const dd = String(now.getDate()).padStart(2, '0')
            const formattedToday = `${yyyy}-${mm}-${dd}`
            const boundary = String(Date.now())
            const timestamp = String(Date.now())
            const ossPath = `Public/Upload/file/run_gyroscope/${boundary.slice(-3)}/${formattedToday}/${timestamp}-${Math.floor(Math.random() * 150)}.txt`
            await this.putOssWithFallback(sts, ossPath, Buffer.from(JSON.stringify(gyrData), 'utf-8'))

            const data = {
                uid: req.uid,
                token: req.token,
                school_id: req.school_id,
                term_id: 0,
                course_id: 0,
                class_id: 0,
                student_num: req.student_id,
                card_id: req.student_id,
                timestamp: this.lepaoTimestamp(),
                version: 1,
                nonce: String(Math.floor(Math.random() * 900000 + 100000)),
                ostype: 5,
                record_id: record_id,
                gyroscope_file: ossPath
            }

            data.sign = dataSign(data)

            return this.request(
                ctx.traceId,
                'bindData',
                this.api('/Run2/gyroscope'),
                data,
                {},
                ctx
            )
        })

        /* ---------------- 提交跑步数据 ---------------- */
        this.register('lepao.bindData', async (req, ctx) => {
            const { pathData } = req

            const distance = Number(Number(pathData.distance || 0).toFixed(2))
            const stepData = generateCadence(distance, pathData.time)
            const stepInfo = JSON.stringify({ interval: 60, list: stepData.cadence_list })

            let points = req.point_data.map(({ address, jingwei, ...rest }) => rest)
            points = JSON.stringify(points)

            const data = {
                uid: req.uid,
                token: req.token,
                school_id: req.school_id,
                term_id: 1,
                course_id: 0,
                class_id: 0,
                student_num: req.student_id,
                card_id: req.student_id,
                timestamp: this.lepaoTimestamp(),
                version: 1,
                nonce: String(Math.floor(Math.random() * 900000 + 100000)),
                ostype: 5,
                game_id: String(req.run_zone_id || 0),
                start_time: req.run_end_time - Number(pathData.time),
                end_time: req.run_end_time,
                distance,
                record_img: "",
                log_data: points,
                file_img: "",
                is_running_area_valid: 1,
                mobileDeviceId: 1,
                mobileModel: req.deviceModel,
                step_info: stepInfo,
                step_num: stepData.total_steps,
                used_time: pathData.time,
                mobileOsVersion: 1,
                record_file: req.record_file
            }

            data.sign = dataSign(data)

            return this.request(
                ctx.traceId,
                'bindData',
                this.api('/Run/stopRunV278'),
                data,
                {},
                ctx
            )
        })
    }

    /* ================= Worker核心 ================= */

    async start() {
        if (this.running) return
        this.running = true

        this.logger.info('Worker 启动中...')

        try {
            this.initHandlers()

            const channel = await mq.getChannel(this.channelName)

            await channel.prefetch(5)

            await assertRunforgeTaskIngress(channel, this.logger)
            await channel.assertQueue(this.resultQueue, {
                durable: true,
                arguments: {
                    'x-max-length': this.maxQueueLength
                }
            })
            await channel.assertQueue(this.deadQueue, {
                durable: true,
                arguments: {
                    'x-max-length': this.maxQueueLength
                }
            })

            await channel.consume(this.taskQueue, async (msg) => {
                if (!msg) return

                let content

                try {
                    content = JSON.parse(msg.content.toString())
                } catch {
                    return channel.ack(msg)
                }

                const { id, type, data, retry = 0 } = content

                const traceId = this.traceId()

                const handler = this.handlers[type]

                if (!handler) {
                    this.log(traceId, 'ERROR', '未知任务', { type })
                    return channel.ack(msg)
                }

                try {
                    const proxyEnabled = await QgProxyManager.isOutboundProxyEnabled()
                    const outboundMode = proxyEnabled ? 'proxy' : 'direct'
                    const result = await this.withTimeout(
                        handler(data, {
                            traceId,
                            channel,
                            taskId: id,
                            outboundMode,
                            retry,
                            maxRetry: this.maxRetry
                        }),
                        type,
                        this.taskTimeoutMs,
                        { code: 'TASK_TIMEOUT', retryable: true }
                    )

                    await this.sendResult(channel, {
                        id,
                        success: true,
                        result
                    })

                    this.log(traceId, 'DONE', `任务完成 ${type}`)
                    channel.ack(msg)

                } catch (err) {
                    this.logErr(traceId, `任务失败 ${type}`, err)

                    if (retry < this.maxRetry && this.isRetryableTaskError(err)) {
                        // 重试
                        await channel.sendToQueue(
                            this.taskQueue,
                            Buffer.from(JSON.stringify({
                                ...content,
                                retry: retry + 1
                            })),
                            { persistent: true }
                        )

                        this.log(traceId, 'RETRY', `重试第${retry + 1}次`)
                    } else {
                        // 死信
                        await channel.sendToQueue(
                            this.deadQueue,
                            Buffer.from(JSON.stringify(content)),
                            { persistent: true }
                        )

                        this.log(traceId, 'DEAD', '进入死信队列')
                    }

                    await this.sendResult(channel, {
                        id,
                        success: false,
                        error: err.code === 'TASK_TIMEOUT' ? '系统繁忙，请稍后再试' : this.maskClientReason(err.message)
                    })

                    channel.ack(msg)
                }
            })

            this.logger.info('RunForge Worker 启动成功')
        } catch (err) {
            this.logger.error('RunForge Worker 启动失败: ' + err.stack)
        }
    }

    async sendResult(channel, data) {
        channel.sendToQueue(
            this.resultQueue,
            Buffer.from(JSON.stringify(data)),
            { persistent: true }
        )
    }

    async stop() {
        this.running = false
        await mq.close()
        this.logger.info('RunForge Worker 已停止')
    }
}

module.exports = Worker