ic-ctbu-backend/lib/QK/TaskScheduler.js

const crypto = require('crypto')
const bcryptjs = require('bcryptjs')
const config = require('../../config.json')
const db = require('../../plugin/DataBase/db')
const Redis = require('../../plugin/DataBase/Redis')
const Logger = require('../Logger')

const TASK_STATUS = {
    PENDING: 'pending',
    ASSIGNED: 'assigned',
    RUNNING: 'running',
    SUCCESS: 'success',
    FAILED: 'failed',
    CANCELLED: 'cancelled'
}

class TaskScheduler {
    constructor(options = {}) {
        this.leaseMs = options.leaseMs || config.qk?.leaseMs || 90 * 1000
        this.heartbeatTtlSeconds = options.heartbeatTtlSeconds || config.qk?.heartbeatTtlSeconds || 45
        this.pullLockTtlSeconds = options.pullLockTtlSeconds || 5
        this.staleTaskMs = options.staleTaskMs || config.qk?.staleTaskMs || 60 * 60 * 1000
        this.memPerSlotMb = options.memPerSlotMb || config.qk?.memPerSlotMb || 3072
        this.memReserveMb = options.memReserveMb || config.qk?.memReserveMb || 1024
        this.maxSlotsCap = options.maxSlotsCap || config.qk?.maxSlotsCap || 10
        this.logger = options.logger || new Logger()
    }

    calculateMaxSlots(profile = {}) {
        const freeMb = Math.max(0, Number(profile.free_mem_mb) || 0)
        const totalMb = Math.max(0, Number(profile.total_mem_mb) || 0)
        const threads = Math.max(1, Number(profile.cpu_threads) || 1)
        const allocatableFreeMb = Math.max(0, freeMb - this.memReserveMb)
        const allocatableTotalMb = Math.max(0, totalMb - this.memReserveMb)
        const byFreeMem = Math.floor(allocatableFreeMb / this.memPerSlotMb)
        const byTotalMem = Math.floor(allocatableTotalMb / this.memPerSlotMb)
        const byCpu = Math.floor(threads * 0.8)
        const fallbackMem = totalMb > 0 ? byTotalMem : 1
        const byMem = freeMb > 0 ? Math.min(byFreeMem, byTotalMem) : fallbackMem
        return Math.max(1, Math.min(50, this.maxSlotsCap, byMem, byCpu))
    }

    resolveClientMaxSlots(client, payload = {}) {
        const fromPayload = this.calculateMaxSlots({
            free_mem_mb: payload.free_mem_mb ?? client.free_mem_mb,
            total_mem_mb: payload.total_mem_mb ?? client.total_mem_mb,
            cpu_threads: payload.cpu_threads ?? client.cpu_threads
        })
        const reported = Number(payload.max_slots || client.max_slots || 0)
        if (!reported) return fromPayload
        return Math.max(1, Math.min(reported, fromPayload))
    }

    getClientAvailableSlots(client, payload = {}) {
        const maxSlots = this.resolveClientMaxSlots(client, payload)
        const currentSlots = Math.max(0, Number(payload.current_slots ?? client.current_slots ?? 0))
        return Math.max(0, maxSlots - currentSlots)
    }

    async countOnlineClients() {
        const now = Date.now()
        const threshold = now - this.heartbeatTtlSeconds * 1000
        const rows = await db.query(
            `SELECT COUNT(*) AS total FROM qk_client
            WHERE enabled = 1 AND online = 1
            AND last_heartbeat_at IS NOT NULL AND last_heartbeat_at >= ?`,
            [threshold]
        )
        return Number(rows?.[0]?.total || 0)
    }

    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
        })
    }

    sanitizeForLog(payload) {
        if (!payload || typeof payload !== 'object') {
            return payload
        }
        const copy = Array.isArray(payload) ? [...payload] : { ...payload }
        for (const key of ['password', 'pass', 'password_enc', 'client_secret']) {
            if (key in copy) {
                copy[key] = '***'
            }
        }
        return copy
    }

    buildLogPrefix(tag, ctx = {}) {
        const parts = ['[QK]', `[${tag}]`]
        if (ctx.taskId) parts.push(`[taskId=${ctx.taskId}]`)
        if (ctx.clientId) parts.push(`[clientId=${ctx.clientId}]`)
        if (ctx.uuid) parts.push(`[uuid=${ctx.uuid}]`)
        return parts.join('')
    }

    logInfo(tag, message, ctx = {}, data = null) {
        const prefix = this.buildLogPrefix(tag, ctx)
        const suffix = data != null ? ` ${this.safeStringify(this.sanitizeForLog(data))}` : ''
        this.logger.info(`${prefix} ${message}${suffix}`)
    }

    logWarn(tag, message, ctx = {}, data = null) {
        const prefix = this.buildLogPrefix(tag, ctx)
        const suffix = data != null ? ` ${this.safeStringify(this.sanitizeForLog(data))}` : ''
        this.logger.warn(`${prefix} ${message}${suffix}`)
    }

    logError(tag, message, ctx = {}, err = null) {
        const prefix = this.buildLogPrefix(tag, ctx)
        const suffix = err ? ` ${err.stack || err}` : ''
        this.logger.error(`${prefix} ${message}${suffix}`)
    }

    getPasswordKey() {
        const source = process.env.QK_PASSWORD_KEY || config.qk?.passwordAesKey || config.database?.password || 'runforge-qk-default-key'
        return crypto.createHash('sha256').update(String(source)).digest()
    }

    encryptPassword(password) {
        const iv = crypto.randomBytes(16)
        const cipher = crypto.createCipheriv('aes-256-cbc', this.getPasswordKey(), iv)
        let encrypted = cipher.update(String(password), 'utf8', 'base64')
        encrypted += cipher.final('base64')
        return `${iv.toString('base64')}:${encrypted}`
    }

    decryptPassword(encrypted) {
        const [ivText, payload] = String(encrypted || '').split(':')
        if (!ivText || !payload) {
            return ''
        }
        const decipher = crypto.createDecipheriv('aes-256-cbc', this.getPasswordKey(), Buffer.from(ivText, 'base64'))
        let decrypted = decipher.update(payload, 'base64', 'utf8')
        decrypted += decipher.final('utf8')
        return decrypted
    }

    normalizeArray(value) {
        if (Array.isArray(value)) {
            return value.map(item => String(item).trim()).filter(Boolean)
        }
        if (typeof value === 'string') {
            const trimmed = value.trim()
            if (!trimmed) {
                return []
            }
            try {
                const parsed = JSON.parse(trimmed)
                if (Array.isArray(parsed)) {
                    return parsed.map(item => String(item).trim()).filter(Boolean)
                }
            } catch (_) {
                return trimmed.split(/[\n,，]/).map(item => item.trim()).filter(Boolean)
            }
        }
        return []
    }

    serializeTask(row, includeSecret = false) {
        const result = { ...row }
        result.enable_ggxxk = Number(result.enable_ggxxk) === 1
        result.courses = this.normalizeArray(result.courses)
        result.course_groups = this.normalizeArray(result.course_groups)
        if (typeof result.result_json === 'string' && result.result_json) {
            try {
                result.result_json = JSON.parse(result.result_json)
            } catch (_) {}
        }
        if (includeSecret) {
            result.password = this.decryptPassword(result.password_enc)
        }
        delete result.password_enc
        return result
    }

    async logTask(taskId, clientId, event, message = '', payload = null) {
        const sql = 'INSERT INTO qk_task_log (task_id, client_id, event, message, payload_json, create_time) VALUES (?, ?, ?, ?, ?, ?)'
        await db.query(sql, [
            taskId,
            clientId || null,
            event,
            message || '',
            payload ? JSON.stringify(payload) : null,
            Date.now()
        ])
        this.logInfo('taskLog', message || event, { taskId, clientId }, payload ? { event, ...this.sanitizeForLog(payload) } : { event })
    }

    async createTask(uuid, payload) {
        const courses = this.normalizeArray(payload.courses || payload.COURSES)
        const courseGroups = this.normalizeArray(payload.course_groups || payload.COURSE_GROUPS)
        const intervalMs = Number(payload.interval_ms || payload.INTERVAL_MS || 500)
        if (!courses.length && !courseGroups.length) {
            throw new Error('至少需要填写一门课程或一个课程分组')
        }
        if (!Number.isFinite(intervalMs) || intervalMs < 200 || intervalMs > 10000) {
            throw new Error('抢课间隔需在 200-10000ms 之间')
        }
        const time = Date.now()
        const sql = `INSERT INTO qk_task
            (create_user, name, jx0502zbid, student_num, password_enc, courses, course_groups, enable_ggxxk, interval_ms, status, create_time, update_time)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
        const result = await db.query(sql, [
            uuid,
            payload.name,
            payload.jx0502zbid || payload.id,
            payload.student_num || payload.user,
            this.encryptPassword(payload.password || payload.pass),
            JSON.stringify(courses),
            JSON.stringify(courseGroups),
            payload.enable_ggxxk || payload.ENABLE_GGXXK ? 1 : 0,
            intervalMs,
            TASK_STATUS.PENDING,
            time,
            time
        ])
        if (!result || result.affectedRows <= 0) {
            throw new Error('创建抢课任务失败')
        }
        await this.logTask(result.insertId, null, 'created', '用户提交抢课任务')
        this.logInfo('createTask', '抢课任务已创建', { taskId: result.insertId, uuid }, {
            name: payload.name,
            student_num: payload.student_num || payload.user,
            jx0502zbid: payload.jx0502zbid || payload.id,
            courses_count: courses.length,
            course_groups_count: courseGroups.length,
            interval_ms: intervalMs,
            enable_ggxxk: !!(payload.enable_ggxxk || payload.ENABLE_GGXXK)
        })
        return result.insertId
    }

    async updateTask(uuid, taskId, payload) {
        const rows = await db.query('SELECT status FROM qk_task WHERE id = ? AND create_user = ?', [taskId, uuid])
        if (!rows || rows.length === 0) {
            throw new Error('任务不存在')
        }
        if (![TASK_STATUS.PENDING, TASK_STATUS.FAILED, TASK_STATUS.CANCELLED].includes(rows[0].status)) {
            throw new Error('任务已被客户端领取，暂不能修改')
        }
        const courses = this.normalizeArray(payload.courses || payload.COURSES)
        const courseGroups = this.normalizeArray(payload.course_groups || payload.COURSE_GROUPS)
        const intervalMs = Number(payload.interval_ms || payload.INTERVAL_MS || 500)
        if (!courses.length && !courseGroups.length) {
            throw new Error('至少需要填写一门课程或一个课程分组')
        }
        const passwordSql = payload.password || payload.pass ? ', password_enc = ?' : ''
        const params = [
            payload.name,
            payload.jx0502zbid || payload.id,
            payload.student_num || payload.user,
            JSON.stringify(courses),
            JSON.stringify(courseGroups),
            payload.enable_ggxxk || payload.ENABLE_GGXXK ? 1 : 0,
            intervalMs,
            TASK_STATUS.PENDING,
            Date.now()
        ]
        if (passwordSql) {
            params.splice(3, 0, this.encryptPassword(payload.password || payload.pass))
        }
        params.push(taskId, uuid)
        const sql = `UPDATE qk_task SET name = ?, jx0502zbid = ?, student_num = ?${passwordSql}, courses = ?, course_groups = ?, enable_ggxxk = ?, interval_ms = ?, status = ?, update_time = ?, assigned_client_id = NULL, lease_expire_at = NULL, result_json = NULL, error_msg = NULL, finished_time = NULL WHERE id = ? AND create_user = ?`
        const result = await db.query(sql, params)
        if (!result || result.affectedRows <= 0) {
            throw new Error('更新抢课任务失败')
        }
        await this.logTask(taskId, null, 'updated', '用户更新抢课任务')
        this.logInfo('updateTask', '抢课任务已更新并重新进入待分配队列', { taskId, uuid }, {
            name: payload.name,
            student_num: payload.student_num || payload.user,
            courses_count: courses.length,
            course_groups_count: courseGroups.length,
            interval_ms: intervalMs,
            password_changed: !!(payload.password || payload.pass)
        })
    }

    async listUserTasks(uuid) {
        const rows = await db.query('SELECT * FROM qk_task WHERE create_user = ? ORDER BY create_time DESC', [uuid])
        return (rows || []).map(row => this.serializeTask(row))
    }

    async getTaskDetail(uuid, taskId) {
        const rows = await db.query('SELECT * FROM qk_task WHERE id = ? AND create_user = ?', [taskId, uuid])
        if (!rows || rows.length === 0) {
            return null
        }
        const logs = await db.query('SELECT client_id, event, message, payload_json, create_time FROM qk_task_log WHERE task_id = ? ORDER BY create_time DESC LIMIT 100', [taskId])
        return {
            task: this.serializeTask(rows[0]),
            logs: (logs || []).map(log => {
                if (typeof log.payload_json === 'string' && log.payload_json) {
                    try {
                        log.payload_json = JSON.parse(log.payload_json)
                    } catch (_) {}
                }
                return log
            })
        }
    }

    async cancelTask(uuid, taskId) {
        const result = await db.query(
            'UPDATE qk_task SET status = ?, update_time = ?, finished_time = ? WHERE id = ? AND create_user = ? AND status IN (?, ?)',
            [TASK_STATUS.CANCELLED, Date.now(), Date.now(), taskId, uuid, TASK_STATUS.PENDING, TASK_STATUS.FAILED]
        )
        if (!result || result.affectedRows <= 0) {
            throw new Error('任务不存在或当前状态不可取消')
        }
        await this.logTask(taskId, null, 'cancelled', '用户取消抢课任务')
        this.logInfo('cancelTask', '用户已取消抢课任务', { taskId, uuid })
    }

    async getAdminTask(taskId) {
        const rows = await db.query(
            `SELECT t.*, u.username, u.avatar
            FROM qk_task t
            LEFT JOIN users u ON u.uuid COLLATE utf8mb4_general_ci = t.create_user COLLATE utf8mb4_general_ci
            WHERE t.id = ?`,
            [taskId]
        )
        if (!rows || rows.length === 0) {
            return null
        }
        return this.serializeTask(rows[0], true)
    }

    async decrementClientSlots(clientId, now = Date.now()) {
        if (!clientId) return
        await db.query(
            'UPDATE qk_client SET current_slots = IF(current_slots > 0, current_slots - 1, 0), update_time = ? WHERE client_id = ?',
            [now, clientId]
        )
    }

    async adminUpdateTask(taskId, payload) {
        const rows = await db.query('SELECT * FROM qk_task WHERE id = ?', [taskId])
        if (!rows || rows.length === 0) {
            throw new Error('任务不存在')
        }
        const task = rows[0]
        if (task.status === TASK_STATUS.SUCCESS) {
            throw new Error('已成功的任务不可编辑')
        }
        const courses = this.normalizeArray(payload.courses || payload.COURSES)
        const courseGroups = this.normalizeArray(payload.course_groups || payload.COURSE_GROUPS)
        const intervalMs = Number(payload.interval_ms || payload.INTERVAL_MS || task.interval_ms || 500)
        if (!courses.length && !courseGroups.length) {
            throw new Error('至少需要填写一门课程或一个课程分组')
        }
        if (!Number.isFinite(intervalMs) || intervalMs < 200 || intervalMs > 10000) {
            throw new Error('抢课间隔需在 200-10000ms 之间')
        }
        const now = Date.now()
        const wasAssigned = [TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING].includes(task.status)
        if (wasAssigned) {
            await this.decrementClientSlots(task.assigned_client_id, now)
        }
        const passwordSql = payload.password || payload.pass ? ', password_enc = ?' : ''
        const params = [
            payload.name,
            payload.jx0502zbid || payload.id,
            payload.student_num || payload.user,
            JSON.stringify(courses),
            JSON.stringify(courseGroups),
            payload.enable_ggxxk || payload.ENABLE_GGXXK ? 1 : 0,
            intervalMs,
            TASK_STATUS.PENDING,
            now
        ]
        if (passwordSql) {
            params.splice(3, 0, this.encryptPassword(payload.password || payload.pass))
        }
        params.push(taskId)
        const sql = `UPDATE qk_task SET name = ?, jx0502zbid = ?, student_num = ?${passwordSql}, courses = ?, course_groups = ?, enable_ggxxk = ?, interval_ms = ?, status = ?, update_time = ?, assigned_client_id = NULL, lease_expire_at = NULL, assigned_at = NULL, exclude_client_id = NULL, result_json = NULL, error_msg = NULL, finished_time = NULL WHERE id = ?`
        const result = await db.query(sql, params)
        if (!result || result.affectedRows <= 0) {
            throw new Error('更新抢课任务失败')
        }
        await this.logTask(taskId, task.assigned_client_id, 'admin_updated', wasAssigned ? '管理员更新任务并收回重新排队' : '管理员更新抢课任务')
        this.logInfo('adminUpdateTask', '管理员已更新抢课任务', { taskId }, {
            name: payload.name,
            student_num: payload.student_num || payload.user,
            released_from_client: wasAssigned ? task.assigned_client_id : null
        })
    }

    async adminCancelTask(taskId) {
        const rows = await db.query('SELECT id, status, assigned_client_id FROM qk_task WHERE id = ?', [taskId])
        if (!rows || rows.length === 0) {
            throw new Error('任务不存在')
        }
        const task = rows[0]
        if (![TASK_STATUS.PENDING, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING, TASK_STATUS.FAILED].includes(task.status)) {
            throw new Error('当前状态不可取消')
        }
        const now = Date.now()
        if ([TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING].includes(task.status)) {
            await this.decrementClientSlots(task.assigned_client_id, now)
        }
        const result = await db.query(
            `UPDATE qk_task SET status = ?, assigned_client_id = NULL, lease_expire_at = NULL, assigned_at = NULL,
            exclude_client_id = NULL, update_time = ?, finished_time = ? WHERE id = ? AND status IN (?, ?, ?, ?)`,
            [TASK_STATUS.CANCELLED, now, now, taskId, TASK_STATUS.PENDING, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING, TASK_STATUS.FAILED]
        )
        if (!result || result.affectedRows <= 0) {
            throw new Error('取消抢课任务失败')
        }
        await this.logTask(taskId, task.assigned_client_id, 'admin_cancelled', '管理员取消抢课任务')
        this.logInfo('adminCancelTask', '管理员已取消抢课任务', { taskId })
    }

    async adminRetryTask(taskId) {
        const rows = await db.query('SELECT id, status FROM qk_task WHERE id = ?', [taskId])
        if (!rows || rows.length === 0) {
            throw new Error('任务不存在')
        }
        if (![TASK_STATUS.FAILED, TASK_STATUS.CANCELLED].includes(rows[0].status)) {
            throw new Error('仅失败或已取消的任务可重试')
        }
        const now = Date.now()
        const result = await db.query(
            `UPDATE qk_task SET status = ?, assigned_client_id = NULL, lease_expire_at = NULL, assigned_at = NULL,
            exclude_client_id = NULL, result_json = NULL, error_msg = NULL, finished_time = NULL, update_time = ?
            WHERE id = ? AND status IN (?, ?)`,
            [TASK_STATUS.PENDING, now, taskId, TASK_STATUS.FAILED, TASK_STATUS.CANCELLED]
        )
        if (!result || result.affectedRows <= 0) {
            throw new Error('重试抢课任务失败')
        }
        await this.logTask(taskId, null, 'admin_retry', '管理员将任务重新加入待分配队列')
        this.logInfo('adminRetryTask', '管理员已重试抢课任务', { taskId })
    }

    async authenticateClient(clientId, clientSecret) {
        if (!clientId || !clientSecret) {
            this.logWarn('authClient', '客户端认证失败：缺少凭证', { clientId: clientId || 'unknown' })
            return null
        }
        const rows = await db.query('SELECT * FROM qk_client WHERE client_id = ? AND enabled = 1', [clientId])
        if (!rows || rows.length !== 1) {
            this.logWarn('authClient', '客户端认证失败：客户端不存在或已禁用', { clientId })
            return null
        }
        if (!bcryptjs.compareSync(String(clientSecret), rows[0].client_secret_hash)) {
            this.logWarn('authClient', '客户端认证失败：密钥不匹配', { clientId })
            return null
        }
        return rows[0]
    }

    async enrollOrAuthenticateClient(clientId, clientSecret, payload = {}) {
        const existing = await this.authenticateClient(clientId, clientSecret)
        if (existing) {
            return existing
        }
        if (!clientId || !clientSecret || !String(clientId).startsWith('qk-cli-')) {
            throw new Error('客户端凭证无效')
        }
        const rows = await db.query('SELECT * FROM qk_client WHERE client_id = ?', [clientId])
        if (rows && rows.length > 0) {
            throw new Error('客户端凭证无效')
        }
        const time = Date.now()
        const label = payload.label || payload.hostname || `auto-${clientId}`
        try {
            const result = await db.query(
                'INSERT INTO qk_client (client_id, client_secret_hash, label, create_time, update_time) VALUES (?, ?, ?, ?, ?)',
                [clientId, bcryptjs.hashSync(String(clientSecret), 10), label, time, time]
            )
            if (!result || result.affectedRows <= 0) {
                throw new Error('客户端自动注册失败')
            }
        } catch (err) {
            if (err?.code === 'ER_DUP_ENTRY') {
                const raced = await this.authenticateClient(clientId, clientSecret)
                if (raced) {
                    return raced
                }
            }
            throw err
        }
        const created = await db.query('SELECT * FROM qk_client WHERE client_id = ? AND enabled = 1', [clientId])
        if (!created || created.length !== 1) {
            throw new Error('客户端自动注册失败')
        }
        this.logInfo('enrollClient', '抢课客户端已自动注册', { clientId }, { label })
        return created[0]
    }

    async registerClient(clientId, clientSecret, payload = {}) {
        const client = await this.enrollOrAuthenticateClient(clientId, clientSecret, payload)
        const maxSlots = this.resolveClientMaxSlots(client, payload)
        const currentSlots = Math.max(0, Number(payload.current_slots || 0))
        const time = Date.now()
        await db.query(
            'UPDATE qk_client SET label = COALESCE(NULLIF(?, \'\'), label), max_slots = ?, current_slots = ?, hostname = ?, os_username = ?, cpu_model = ?, cpu_threads = ?, total_mem_mb = ?, free_mem_mb = ?, platform = ?, last_heartbeat_at = ?, online = 1, update_time = ? WHERE client_id = ?',
            [
                payload.label || '',
                maxSlots,
                currentSlots,
                payload.hostname || null,
                payload.os_username || null,
                payload.cpu_model || null,
                payload.cpu_threads || null,
                payload.total_mem_mb || null,
                payload.free_mem_mb || null,
                payload.platform || null,
                time,
                time,
                clientId
            ]
        )
        await Redis.set(`qk:client:hb:${clientId}`, String(time), { EX: this.heartbeatTtlSeconds })
        this.logInfo('registerClient', '抢课客户端已注册/上线', { clientId }, {
            label: payload.label || client.label,
            max_slots: maxSlots,
            current_slots: currentSlots,
            hostname: payload.hostname,
            platform: payload.platform,
            cpu_threads: payload.cpu_threads,
            total_mem_mb: payload.total_mem_mb,
            free_mem_mb: payload.free_mem_mb
        })
        return { client_id: clientId, max_slots: maxSlots, current_slots: currentSlots }
    }

    async heartbeat(clientId, clientSecret, payload = {}) {
        const client = await this.authenticateClient(clientId, clientSecret)
        if (!client) {
            throw new Error('客户端凭证无效')
        }
        const runningTasks = Array.isArray(payload.running_tasks) ? payload.running_tasks.map(Number).filter(Boolean) : []
        const currentSlots = Math.max(0, Number(payload.current_slots ?? runningTasks.length))
        const maxSlots = this.resolveClientMaxSlots(client, payload)
        const now = Date.now()
        await db.query(
            'UPDATE qk_client SET max_slots = ?, current_slots = ?, hostname = ?, os_username = ?, cpu_model = ?, cpu_threads = ?, total_mem_mb = ?, free_mem_mb = ?, platform = ?, last_heartbeat_at = ?, online = 1, update_time = ? WHERE client_id = ?',
            [
                maxSlots,
                currentSlots,
                payload.hostname || null,
                payload.os_username || null,
                payload.cpu_model || null,
                payload.cpu_threads || null,
                payload.total_mem_mb || null,
                payload.free_mem_mb || null,
                payload.platform || null,
                now,
                now,
                clientId
            ]
        )
        await Redis.set(`qk:client:hb:${clientId}`, String(now), { EX: this.heartbeatTtlSeconds })
        await Redis.set(`qk:client:slots:${clientId}`, String(currentSlots), { EX: this.heartbeatTtlSeconds })
        if (runningTasks.length > 0) {
            const leaseExpireAt = now + this.leaseMs
            const placeholders = runningTasks.map(() => '?').join(',')
            await db.query(
                `UPDATE qk_task SET status = ?, lease_expire_at = ?, update_time = ? WHERE assigned_client_id = ? AND id IN (${placeholders}) AND status IN (?, ?)`,
                [TASK_STATUS.RUNNING, leaseExpireAt, now, clientId, ...runningTasks, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
            )
            this.logInfo('heartbeat', '客户端心跳续租运行中任务', { clientId }, {
                running_tasks: runningTasks,
                lease_expire_at: leaseExpireAt,
                current_slots: currentSlots,
                max_slots: maxSlots
            })
        }
        await this.releaseOrphanedClientTasks(clientId, runningTasks, now)
        return { client_id: clientId, current_slots: currentSlots, max_slots: maxSlots }
    }

    async releaseOrphanedClientTasks(clientId, runningTasks, now = Date.now()) {
        const assigned = await db.query(
            'SELECT id FROM qk_task WHERE assigned_client_id = ? AND status IN (?, ?)',
            [clientId, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
        )
        const runningSet = new Set((runningTasks || []).map(Number).filter(Boolean))
        let released = 0
        for (const row of assigned || []) {
            if (runningSet.has(row.id)) continue
            const result = await db.query(
                'UPDATE qk_task SET status = ?, assigned_client_id = NULL, lease_expire_at = NULL, assigned_at = NULL, update_time = ? WHERE id = ? AND assigned_client_id = ? AND status IN (?, ?)',
                [TASK_STATUS.PENDING, now, row.id, clientId, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
            )
            if (result && result.affectedRows > 0) {
                released += 1
                await this.logTask(row.id, clientId, 'released', '客户端未继续执行，任务已释放回队列')
                this.logInfo('releaseOrphaned', '释放未在运行的已分配任务', { taskId: row.id, clientId })
            }
        }
        return released
    }

    async reclaimTasks(clientId, clientSecret) {
        const client = await this.authenticateClient(clientId, clientSecret)
        if (!client) {
            throw new Error('客户端凭证无效')
        }
        const now = Date.now()
        const leaseExpireAt = now + this.leaseMs
        const rows = await db.query(
            'SELECT * FROM qk_task WHERE assigned_client_id = ? AND status IN (?, ?) ORDER BY create_time ASC',
            [clientId, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
        )
        if (!rows || rows.length === 0) {
            return []
        }
        for (const row of rows) {
            await db.query(
                'UPDATE qk_task SET status = ?, lease_expire_at = ?, update_time = ? WHERE id = ? AND assigned_client_id = ? AND status IN (?, ?)',
                [TASK_STATUS.RUNNING, leaseExpireAt, now, row.id, clientId, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
            )
            await this.logTask(row.id, clientId, 'reclaimed', '客户端重连后回收任务继续执行')
        }
        this.logInfo('reclaimTasks', `客户端回收 ${rows.length} 个进行中任务`, { clientId }, {
            task_ids: rows.map(row => row.id),
            lease_expire_at: leaseExpireAt
        })
        return rows.map(row => this.serializeTask({
            ...row,
            assigned_client_id: clientId,
            lease_expire_at: leaseExpireAt,
            status: TASK_STATUS.RUNNING
        }, true))
    }

    async releaseTasks(clientId, clientSecret, payload = {}) {
        const client = await this.authenticateClient(clientId, clientSecret)
        if (!client) {
            throw new Error('客户端凭证无效')
        }
        const now = Date.now()
        const taskIds = Array.isArray(payload.task_ids)
            ? payload.task_ids.map(Number).filter(Boolean)
            : []
        let rows = []
        if (taskIds.length > 0) {
            const placeholders = taskIds.map(() => '?').join(',')
            rows = await db.query(
                `SELECT id FROM qk_task WHERE assigned_client_id = ? AND id IN (${placeholders}) AND status IN (?, ?)`,
                [clientId, ...taskIds, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
            )
        } else {
            rows = await db.query(
                'SELECT id FROM qk_task WHERE assigned_client_id = ? AND status IN (?, ?)',
                [clientId, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
            )
        }
        let released = 0
        for (const row of rows || []) {
            const result = await db.query(
                'UPDATE qk_task SET status = ?, assigned_client_id = NULL, lease_expire_at = NULL, assigned_at = NULL, update_time = ? WHERE id = ? AND assigned_client_id = ? AND status IN (?, ?)',
                [TASK_STATUS.PENDING, now, row.id, clientId, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
            )
            if (result && result.affectedRows > 0) {
                released += 1
                await this.logTask(row.id, clientId, 'released', '客户端主动释放任务')
            }
        }
        if (released > 0) {
            this.logInfo('releaseTasks', `客户端释放 ${released} 个任务`, { clientId }, {
                task_ids: (rows || []).map(row => row.id)
            })
        }
        return { released, task_ids: (rows || []).map(row => row.id) }
    }

    async pullTasks(clientId, clientSecret, count) {
        const client = await this.authenticateClient(clientId, clientSecret)
        if (!client) {
            throw new Error('客户端凭证无效')
        }
        const availableSlots = this.getClientAvailableSlots(client)
        const safeCount = Math.max(0, Math.min(50, Number(count || 0), availableSlots))
        if (safeCount <= 0) {
            this.logWarn('pullTasks', '客户端无可用槽位或拉取数量无效，已忽略', { clientId }, {
                count,
                available_slots: availableSlots,
                max_slots: this.resolveClientMaxSlots(client),
                current_slots: client.current_slots,
                free_mem_mb: client.free_mem_mb
            })
            return []
        }
        const lockKey = `qk:pull:lock:${clientId}`
        const locked = await Redis.set(lockKey, '1', { NX: true, EX: this.pullLockTtlSeconds })
        if (!locked) {
            this.logWarn('pullTasks', '拉取任务被并发锁拦截，本次跳过', { clientId }, { count: safeCount })
            return []
        }
        const conn = await db.connect()
        try {
            await conn.beginTransaction()
            const [rows] = await conn.execute(
                `SELECT * FROM qk_task
                WHERE status = ?
                AND (exclude_client_id IS NULL OR exclude_client_id <> ?)
                ORDER BY create_time ASC LIMIT ${safeCount} FOR UPDATE`,
                [TASK_STATUS.PENDING, clientId]
            )
            const now = Date.now()
            const leaseExpireAt = now + this.leaseMs
            for (const row of rows) {
                await conn.execute(
                    'UPDATE qk_task SET status = ?, assigned_client_id = ?, lease_expire_at = ?, assigned_at = ?, exclude_client_id = NULL, update_time = ? WHERE id = ?',
                    [TASK_STATUS.ASSIGNED, clientId, leaseExpireAt, now, now, row.id]
                )
            }
            await conn.commit()
            if (rows.length > 0) {
                await db.query(
                    'UPDATE qk_client SET current_slots = current_slots + ?, update_time = ? WHERE client_id = ?',
                    [rows.length, now, clientId]
                )
            }
            for (const row of rows) {
                await this.logTask(row.id, clientId, 'assigned', '任务已分配给客户端')
            }
            if (rows.length > 0) {
                this.logInfo('pullTasks', `已分配 ${rows.length} 个抢课任务`, { clientId }, {
                    task_ids: rows.map(row => row.id),
                    lease_expire_at: leaseExpireAt,
                    requested_count: safeCount
                })
            }
            return rows.map(row => this.serializeTask({ ...row, assigned_client_id: clientId, lease_expire_at: leaseExpireAt, status: TASK_STATUS.ASSIGNED }, true))
        } catch (err) {
            await conn.rollback()
            this.logError('pullTasks', '拉取并分配任务失败，事务已回滚', { clientId }, err)
            throw err
        } finally {
            await Redis.del(lockKey)
        }
    }

    async reportResult(clientId, clientSecret, payload = {}) {
        const client = await this.authenticateClient(clientId, clientSecret)
        if (!client) {
            throw new Error('客户端凭证无效')
        }
        const taskId = Number(payload.task_id || payload.id)
        const success = payload.success === true || payload.status === TASK_STATUS.SUCCESS
        const status = success ? TASK_STATUS.SUCCESS : TASK_STATUS.FAILED
        const now = Date.now()
        const resultJson = payload.result ? JSON.stringify(payload.result) : JSON.stringify({
            course: payload.course || '',
            message: payload.message || ''
        })
        const result = await db.query(
            'UPDATE qk_task SET status = ?, result_json = ?, error_msg = ?, update_time = ?, finished_time = ?, lease_expire_at = NULL WHERE id = ? AND assigned_client_id = ? AND status IN (?, ?)',
            [status, resultJson, success ? null : (payload.error_msg || payload.message || '抢课失败'), now, now, taskId, clientId, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
        )
        if (!result || result.affectedRows <= 0) {
            this.logWarn('reportResult', '任务结果上报被拒绝：任务不存在或不属于当前客户端', { taskId, clientId }, {
                success,
                status
            })
            throw new Error('任务不存在或不属于当前客户端')
        }
        await db.query(
            `UPDATE qk_client SET current_slots = IF(current_slots > 0, current_slots - 1, 0), total_completed = total_completed + 1, total_success = total_success + ?, update_time = ? WHERE client_id = ?`,
            [success ? 1 : 0, now, clientId]
        )
        await this.logTask(taskId, clientId, success ? 'grab_success' : 'grab_fail', payload.message || payload.error_msg || '', payload.result || payload)
        this.logInfo('reportResult', success ? '抢课成功' : '抢课失败', { taskId, clientId }, {
            success,
            message: payload.message || payload.error_msg || '',
            course: payload.course || payload.result?.course || '',
            result: payload.result || null
        })
        return { task_id: taskId, status }
    }

    async listClients() {
        const rows = await db.query('SELECT id, client_id, label, enabled, max_slots, current_slots, hostname, os_username, cpu_model, cpu_threads, total_mem_mb, free_mem_mb, platform, last_heartbeat_at, online, total_completed, total_success, create_time, update_time FROM qk_client ORDER BY update_time DESC')
        return rows || []
    }

    async deleteClient(clientId) {
        const result = await db.query('UPDATE qk_client SET enabled = 0, online = 0, update_time = ? WHERE client_id = ?', [Date.now(), clientId])
        if (!result || result.affectedRows <= 0) {
            throw new Error('客户端不存在')
        }
        this.logInfo('deleteClient', '抢课客户端已禁用', { clientId })
    }

    async listAdminTasks(filters = {}) {
        const pagesize = Math.max(1, Math.min(100, Number(filters.pagesize || 20)))
        const current = Math.max(1, Number(filters.current || 1))
        const where = ['1 = 1']
        const params = []
        const countParams = []
        if (filters.status) {
            where.push('t.status = ?')
            params.push(filters.status)
            countParams.push(filters.status)
        }
        if (filters.client_id) {
            where.push('t.assigned_client_id = ?')
            params.push(filters.client_id)
            countParams.push(filters.client_id)
        }
        if (filters.student_num) {
            where.push('t.student_num LIKE ?')
            params.push(`%${filters.student_num}%`)
            countParams.push(`%${filters.student_num}%`)
        }
        if (filters.username) {
            where.push('u.username COLLATE utf8mb4_general_ci LIKE (CONVERT(? USING utf8mb4) COLLATE utf8mb4_general_ci)')
            params.push(`%${filters.username}%`)
            countParams.push(`%${filters.username}%`)
        }
        const whereSql = where.join(' AND ')
        const offset = (current - 1) * pagesize
        const countRows = await db.query(
            `SELECT COUNT(*) AS total FROM qk_task t
            LEFT JOIN users u ON u.uuid COLLATE utf8mb4_general_ci = t.create_user COLLATE utf8mb4_general_ci
            WHERE ${whereSql}`,
            countParams
        )
        const rows = await db.query(
            `SELECT t.*, u.username, u.avatar
            FROM qk_task t
            LEFT JOIN users u ON u.uuid COLLATE utf8mb4_general_ci = t.create_user COLLATE utf8mb4_general_ci
            WHERE ${whereSql}
            ORDER BY t.create_time DESC
            LIMIT ${pagesize} OFFSET ${offset}`,
            params
        )
        return {
            list: (rows || []).map(row => this.serializeTask(row, true)),
            total: countRows?.[0]?.total || 0,
            current,
            pagesize
        }
    }

    async reassignStaleRunningTasks() {
        const onlineCount = await this.countOnlineClients()
        if (onlineCount <= 1) {
            return 0
        }

        const now = Date.now()
        const staleBefore = now - this.staleTaskMs
        const rows = await db.query(
            `SELECT id, assigned_client_id FROM qk_task
            WHERE status IN (?, ?)
            AND assigned_at IS NOT NULL
            AND assigned_at < ?`,
            [TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING, staleBefore]
        )

        let count = 0
        for (const row of rows || []) {
            const result = await db.query(
                `UPDATE qk_task
                SET status = ?, assigned_client_id = NULL, lease_expire_at = NULL,
                    assigned_at = NULL, exclude_client_id = ?, update_time = ?
                WHERE id = ? AND status IN (?, ?)`,
                [TASK_STATUS.PENDING, row.assigned_client_id, now, row.id, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
            )
            if (result && result.affectedRows > 0) {
                count += 1
                if (row.assigned_client_id) {
                    await db.query(
                        'UPDATE qk_client SET current_slots = IF(current_slots > 0, current_slots - 1, 0), update_time = ? WHERE client_id = ?',
                        [now, row.assigned_client_id]
                    )
                }
                await this.logTask(row.id, row.assigned_client_id, 'reassigned', '任务超过1小时未成功，已收回并等待分配给其他客户端')
                this.logWarn('reassignStale', '长时间未成功任务已收回', {
                    taskId: row.id,
                    clientId: row.assigned_client_id
                }, { online_clients: onlineCount })
            }
        }
        return count
    }

    async requeueExpiredTasks() {
        const now = Date.now()
        const rows = await db.query(
            'SELECT id, assigned_client_id FROM qk_task WHERE status IN (?, ?) AND lease_expire_at IS NOT NULL AND lease_expire_at < ?',
            [TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING, now]
        )
        let count = 0
        for (const row of rows || []) {
            const result = await db.query(
                'UPDATE qk_task SET status = ?, assigned_client_id = NULL, lease_expire_at = NULL, assigned_at = NULL, update_time = ? WHERE id = ? AND status IN (?, ?)',
                [TASK_STATUS.PENDING, now, row.id, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
            )
            if (result && result.affectedRows > 0) {
                count++
                if (row.assigned_client_id) {
                    await db.query(
                        'UPDATE qk_client SET current_slots = IF(current_slots > 0, current_slots - 1, 0), update_time = ? WHERE client_id = ?',
                        [now, row.assigned_client_id]
                    )
                }
                await this.logTask(row.id, row.assigned_client_id, 'reassigned', '租约过期，任务已重新进入待分配队列')
                this.logWarn('requeueExpired', '租约过期，任务已重新入队', {
                    taskId: row.id,
                    clientId: row.assigned_client_id
                })
            }
        }
        const offlineResult = await db.query(
            'UPDATE qk_client SET online = 0, current_slots = 0, update_time = ? WHERE online = 1 AND (last_heartbeat_at IS NULL OR last_heartbeat_at < ?)',
            [now, now - this.heartbeatTtlSeconds * 1000]
        )
        const offlineCount = offlineResult?.affectedRows || 0
        const staleCount = await this.reassignStaleRunningTasks()
        if (count > 0 || offlineCount > 0 || staleCount > 0) {
            this.logInfo('requeueExpired', '租约巡检完成', {}, {
                expired_tasks: (rows || []).length,
                requeued: count,
                stale_reassigned: staleCount,
                clients_marked_offline: offlineCount
            })
        }
        return count + staleCount
    }
}

module.exports = {
    TaskScheduler,
    TASK_STATUS
}