|
|
@@ -0,0 +1,940 @@
|
|
|
+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
|
|
|
+}
|
