|
|
@@ -1,1843 +0,0 @@
|
|
|
-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 AccessControl = require('../AccessControl')
|
|
|
-
|
|
|
-const TASK_STATUS = {
|
|
|
- PAUSED: 'paused',
|
|
|
- PENDING: 'pending',
|
|
|
- ASSIGNED: 'assigned',
|
|
|
- RUNNING: 'running',
|
|
|
- SUCCESS: 'success',
|
|
|
- FAILED: 'failed',
|
|
|
- CANCELLED: 'cancelled'
|
|
|
-}
|
|
|
-
|
|
|
-const REPORT_LOG_EVENTS = new Set([
|
|
|
- 'request_result',
|
|
|
- 'progress_snapshot',
|
|
|
- 'grab_success',
|
|
|
- 'grab_fail'
|
|
|
-])
|
|
|
-
|
|
|
-class TaskScheduler {
|
|
|
- static _schemaReady = false
|
|
|
-
|
|
|
- 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.slotCapacityFactor = options.slotCapacityFactor || config.qk?.slotCapacityFactor || 1.5
|
|
|
- this.logger = options.logger || new Logger()
|
|
|
- }
|
|
|
-
|
|
|
- async ensureSchema() {
|
|
|
- if (TaskScheduler._schemaReady) {
|
|
|
- return
|
|
|
- }
|
|
|
- try {
|
|
|
- const rows = await db.query("SHOW COLUMNS FROM qk_task LIKE 'auto_relogin'")
|
|
|
- if (!rows || rows.length === 0) {
|
|
|
- await db.query(
|
|
|
- 'ALTER TABLE qk_task ADD COLUMN auto_relogin TINYINT(1) NOT NULL DEFAULT 0 AFTER enable_ggxxk'
|
|
|
- )
|
|
|
- this.logInfo('ensureSchema', '已添加 qk_task.auto_relogin 字段')
|
|
|
- }
|
|
|
- const jwRows = await db.query("SHOW COLUMNS FROM qk_task LIKE 'jw_account_id'")
|
|
|
- if (!jwRows || jwRows.length === 0) {
|
|
|
- await db.query(
|
|
|
- 'ALTER TABLE qk_task ADD COLUMN jw_account_id INT NULL DEFAULT NULL AFTER student_num'
|
|
|
- )
|
|
|
- this.logInfo('ensureSchema', '已添加 qk_task.jw_account_id 字段')
|
|
|
- }
|
|
|
- } catch (err) {
|
|
|
- this.logWarn('ensureSchema', '抢课任务表结构检查失败', {}, err)
|
|
|
- }
|
|
|
- TaskScheduler._schemaReady = true
|
|
|
- }
|
|
|
-
|
|
|
- parseAutoRelogin(payload = {}) {
|
|
|
- if (payload.auto_relogin !== undefined) {
|
|
|
- return payload.auto_relogin === true || Number(payload.auto_relogin) === 1
|
|
|
- }
|
|
|
- if (payload.AUTO_RELOGIN !== undefined) {
|
|
|
- return payload.AUTO_RELOGIN === true || Number(payload.AUTO_RELOGIN) === 1
|
|
|
- }
|
|
|
- return false
|
|
|
- }
|
|
|
-
|
|
|
- extractReportCourseName(payload = {}) {
|
|
|
- if (!payload || typeof payload !== 'object') {
|
|
|
- return ''
|
|
|
- }
|
|
|
- if (payload.course_name) {
|
|
|
- return String(payload.course_name)
|
|
|
- }
|
|
|
- if (payload.course) {
|
|
|
- return String(payload.course)
|
|
|
- }
|
|
|
- if (payload.label) {
|
|
|
- const label = String(payload.label)
|
|
|
- const at = label.indexOf('@')
|
|
|
- if (at > 0) {
|
|
|
- return label.slice(0, at)
|
|
|
- }
|
|
|
- return label
|
|
|
- }
|
|
|
- if (Array.isArray(payload.courses) && payload.courses.length > 0) {
|
|
|
- return payload.courses.join('、')
|
|
|
- }
|
|
|
- return ''
|
|
|
- }
|
|
|
-
|
|
|
- formatReportMessage(message = '', payload = {}) {
|
|
|
- const courseName = this.extractReportCourseName(payload)
|
|
|
- const text = String(message || '').trim()
|
|
|
- if (!courseName) {
|
|
|
- return text
|
|
|
- }
|
|
|
- if (!text) {
|
|
|
- return `[${courseName}]`
|
|
|
- }
|
|
|
- if (text.includes(`[${courseName}]`) || text.startsWith(`${courseName}:`)) {
|
|
|
- return text
|
|
|
- }
|
|
|
- return `[${courseName}] ${text}`
|
|
|
- }
|
|
|
-
|
|
|
- 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
|
|
|
- const base = Math.max(1, Math.min(byMem, byCpu))
|
|
|
- const scaled = Math.max(1, Math.floor(base * this.slotCapacityFactor))
|
|
|
- const cap = Math.max(1, Math.floor(this.maxSlotsCap * this.slotCapacityFactor))
|
|
|
- return Math.max(1, Math.min(50, cap, scaled))
|
|
|
- }
|
|
|
-
|
|
|
- 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(client.current_slots ?? 0))
|
|
|
- return Math.max(0, maxSlots - currentSlots)
|
|
|
- }
|
|
|
-
|
|
|
- async countClientActiveTasks(clientId) {
|
|
|
- if (!clientId) return 0
|
|
|
- const rows = await db.query(
|
|
|
- 'SELECT COUNT(*) AS cnt FROM qk_task WHERE assigned_client_id = ? AND status IN (?, ?)',
|
|
|
- [clientId, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
|
|
|
- )
|
|
|
- return Math.max(0, Number(rows?.[0]?.cnt || 0))
|
|
|
- }
|
|
|
-
|
|
|
- async syncClientSlots(clientId, now = Date.now()) {
|
|
|
- if (!clientId) return 0
|
|
|
- const count = await this.countClientActiveTasks(clientId)
|
|
|
- await db.query(
|
|
|
- 'UPDATE qk_client SET current_slots = ?, update_time = ? WHERE client_id = ?',
|
|
|
- [count, now, clientId]
|
|
|
- )
|
|
|
- try {
|
|
|
- await Redis.set(`qk:client:slots:${clientId}`, String(count), { EX: this.heartbeatTtlSeconds })
|
|
|
- } catch (_) {}
|
|
|
- return count
|
|
|
- }
|
|
|
-
|
|
|
- async findRevokedClientTasks(clientId, runningTaskIds = []) {
|
|
|
- const ids = (runningTaskIds || []).map(Number).filter(Boolean)
|
|
|
- if (!ids.length) return []
|
|
|
- const placeholders = ids.map(() => '?').join(',')
|
|
|
- const rows = await db.query(
|
|
|
- `SELECT id FROM qk_task WHERE id IN (${placeholders})
|
|
|
- AND (assigned_client_id IS NULL OR assigned_client_id <> ? OR status NOT IN (?, ?))`,
|
|
|
- [...ids, clientId, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
|
|
|
- )
|
|
|
- return (rows || []).map(row => row.id)
|
|
|
- }
|
|
|
-
|
|
|
- async getClientAvailableSlotsAsync(client, payload = {}) {
|
|
|
- const maxSlots = this.resolveClientMaxSlots(client, payload)
|
|
|
- const activeCount = await this.countClientActiveTasks(client.client_id)
|
|
|
- return Math.max(0, maxSlots - activeCount)
|
|
|
- }
|
|
|
-
|
|
|
- getInFlightTaskStatuses() {
|
|
|
- return [TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
|
|
|
- }
|
|
|
-
|
|
|
- /** 同一学号在单次拉取中只保留最早创建的一条待分配任务 */
|
|
|
- pickPullableTasks(rows, limit) {
|
|
|
- const picked = []
|
|
|
- const seenStudentNums = new Set()
|
|
|
- for (const row of rows || []) {
|
|
|
- const studentNum = String(row.student_num || '').trim()
|
|
|
- if (!studentNum || seenStudentNums.has(studentNum)) {
|
|
|
- continue
|
|
|
- }
|
|
|
- seenStudentNums.add(studentNum)
|
|
|
- picked.push(row)
|
|
|
- if (picked.length >= limit) {
|
|
|
- break
|
|
|
- }
|
|
|
- }
|
|
|
- return picked
|
|
|
- }
|
|
|
-
|
|
|
- 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 []
|
|
|
- }
|
|
|
-
|
|
|
- countTaskTargets(courses = [], courseGroups = []) {
|
|
|
- return this.normalizeArray(courses).length + this.normalizeArray(courseGroups).length
|
|
|
- }
|
|
|
-
|
|
|
- getMinTaskInterval(totalCount) {
|
|
|
- return totalCount >= 5 ? 500 : 200
|
|
|
- }
|
|
|
-
|
|
|
- validateTaskCoursesAndInterval(courses, courseGroups, intervalMs) {
|
|
|
- const normalizedCourses = this.normalizeArray(courses)
|
|
|
- const normalizedGroups = this.normalizeArray(courseGroups)
|
|
|
- const totalCount = normalizedCourses.length + normalizedGroups.length
|
|
|
- if (totalCount <= 0) {
|
|
|
- throw new Error('至少需要填写一门课程或一个课程分组')
|
|
|
- }
|
|
|
- if (totalCount > 10) {
|
|
|
- throw new Error('每个任务最多选择10门课程或分组')
|
|
|
- }
|
|
|
- const minInterval = this.getMinTaskInterval(totalCount)
|
|
|
- const safeInterval = Number(intervalMs)
|
|
|
- if (!Number.isFinite(safeInterval) || safeInterval < minInterval || safeInterval > 10000) {
|
|
|
- throw new Error(`当前共 ${totalCount} 个抢课目标,间隔需在 ${minInterval}-10000ms 之间`)
|
|
|
- }
|
|
|
- return {
|
|
|
- courses: normalizedCourses,
|
|
|
- courseGroups: normalizedGroups,
|
|
|
- intervalMs: safeInterval,
|
|
|
- totalCount,
|
|
|
- minInterval
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- async resolveTaskAccount(ownerUuid, payload, { requireAccount = true } = {}) {
|
|
|
- const jwAccountId = Number(payload.jw_account_id || payload.jwAccountId)
|
|
|
- if (jwAccountId) {
|
|
|
- const account = await AccessControl.getVerifiedJwAccount(ownerUuid, jwAccountId)
|
|
|
- if (!account) {
|
|
|
- throw new Error('统一身份认证账号不存在或未验证')
|
|
|
- }
|
|
|
- return {
|
|
|
- student_num: account.username,
|
|
|
- password: account.password,
|
|
|
- jw_account_id: account.id
|
|
|
- }
|
|
|
- }
|
|
|
- const studentNum = String(payload.student_num || payload.user || '').trim()
|
|
|
- const password = payload.password || payload.pass
|
|
|
- if (studentNum && password) {
|
|
|
- return {
|
|
|
- student_num: studentNum,
|
|
|
- password,
|
|
|
- jw_account_id: payload.jw_account_id || null
|
|
|
- }
|
|
|
- }
|
|
|
- if (requireAccount) {
|
|
|
- throw new Error('请选择统一身份认证账号')
|
|
|
- }
|
|
|
- return null
|
|
|
- }
|
|
|
-
|
|
|
- async inferJwAccountId(ownerUuid, studentNum, currentId) {
|
|
|
- if (currentId) {
|
|
|
- return Number(currentId)
|
|
|
- }
|
|
|
- const username = String(studentNum || '').trim()
|
|
|
- if (!username) {
|
|
|
- return null
|
|
|
- }
|
|
|
- const rows = await db.query(
|
|
|
- 'SELECT id FROM jw_account WHERE create_user = ? AND username = ? AND state = 1 LIMIT 1',
|
|
|
- [ownerUuid, username]
|
|
|
- )
|
|
|
- return rows?.[0]?.id || null
|
|
|
- }
|
|
|
-
|
|
|
- serializeTask(row, includeSecret = false) {
|
|
|
- const result = { ...row }
|
|
|
- result.enable_ggxxk = Number(result.enable_ggxxk) === 1
|
|
|
- result.auto_relogin = Number(result.auto_relogin) === 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 (_) {}
|
|
|
- }
|
|
|
- result.jx0502zbid = row.batch_jx0502zbid || row.jx0502zbid || ''
|
|
|
- result.batch_name = row.batch_name || ''
|
|
|
- if (row.batch_enabled !== undefined) {
|
|
|
- result.batch_enabled = Number(row.batch_enabled) === 1
|
|
|
- }
|
|
|
- delete result.batch_jx0502zbid
|
|
|
- if (includeSecret) {
|
|
|
- result.password = this.decryptPassword(result.password_enc)
|
|
|
- }
|
|
|
- delete result.password_enc
|
|
|
- return result
|
|
|
- }
|
|
|
-
|
|
|
- taskBatchJoinSql(alias = 't') {
|
|
|
- return `LEFT JOIN qk_batch b ON b.id = ${alias}.batch_id`
|
|
|
- }
|
|
|
-
|
|
|
- taskBatchSelectSql(alias = 't') {
|
|
|
- return `, b.name AS batch_name, b.jx0502zbid AS batch_jx0502zbid, b.enabled AS batch_enabled`
|
|
|
- }
|
|
|
-
|
|
|
- async getBatchById(batchId) {
|
|
|
- const rows = await db.query('SELECT * FROM qk_batch WHERE id = ?', [batchId])
|
|
|
- if (!rows || rows.length === 0) {
|
|
|
- throw new Error('抢课批次不存在')
|
|
|
- }
|
|
|
- return rows[0]
|
|
|
- }
|
|
|
-
|
|
|
- async requireEnabledBatch(batchId) {
|
|
|
- const batch = await this.getBatchById(batchId)
|
|
|
- if (!Number(batch.enabled)) {
|
|
|
- throw new Error('所选抢课批次未启用')
|
|
|
- }
|
|
|
- return batch
|
|
|
- }
|
|
|
-
|
|
|
- resolveTaskBatchId(payload = {}) {
|
|
|
- const batchId = Number(payload.batch_id || payload.batchId)
|
|
|
- if (!batchId) {
|
|
|
- throw new Error('请选择抢课批次')
|
|
|
- }
|
|
|
- return batchId
|
|
|
- }
|
|
|
-
|
|
|
- async releaseAssignedTask(task) {
|
|
|
- return !!(task && [TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING].includes(task.status))
|
|
|
- }
|
|
|
-
|
|
|
- async pauseTasksByBatch(batchId, reason = '批次已停用,任务已收回') {
|
|
|
- const rows = await db.query(
|
|
|
- 'SELECT id, status, assigned_client_id FROM qk_task WHERE batch_id = ? AND status IN (?, ?, ?)',
|
|
|
- [batchId, TASK_STATUS.PENDING, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
|
|
|
- )
|
|
|
- const now = Date.now()
|
|
|
- const affectedClients = new Set()
|
|
|
- let count = 0
|
|
|
- for (const row of rows || []) {
|
|
|
- if (row.assigned_client_id) {
|
|
|
- affectedClients.add(row.assigned_client_id)
|
|
|
- }
|
|
|
- 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 = ? WHERE id = ? AND status IN (?, ?, ?)`,
|
|
|
- [TASK_STATUS.PAUSED, now, row.id, TASK_STATUS.PENDING, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
|
|
|
- )
|
|
|
- if (result && result.affectedRows > 0) {
|
|
|
- count += 1
|
|
|
- await this.logTask(row.id, row.assigned_client_id, 'batch_paused', reason)
|
|
|
- }
|
|
|
- }
|
|
|
- for (const clientId of affectedClients) {
|
|
|
- await this.syncClientSlots(clientId, now)
|
|
|
- }
|
|
|
- if (count > 0) {
|
|
|
- this.logInfo('pauseTasksByBatch', reason, { batchId }, { affected: count })
|
|
|
- }
|
|
|
- return count
|
|
|
- }
|
|
|
-
|
|
|
- async redispatchTasksByBatch(batchId, reason = '批次 ID 已变更,任务已重新进入分配队列') {
|
|
|
- const rows = await db.query(
|
|
|
- 'SELECT id, status, assigned_client_id FROM qk_task WHERE batch_id = ? AND status IN (?, ?, ?)',
|
|
|
- [batchId, TASK_STATUS.PENDING, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
|
|
|
- )
|
|
|
- const now = Date.now()
|
|
|
- let count = 0
|
|
|
- for (const row of rows || []) {
|
|
|
- if ([TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING].includes(row.status)) {
|
|
|
- await this.releaseAssignedTask(row, 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 = ? WHERE id = ? AND status IN (?, ?)`,
|
|
|
- [TASK_STATUS.PENDING, now, row.id, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
|
|
|
- )
|
|
|
- if (result && result.affectedRows > 0) {
|
|
|
- count += 1
|
|
|
- await this.logTask(row.id, row.assigned_client_id, 'batch_redispatch', reason)
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- if (count > 0) {
|
|
|
- this.logInfo('redispatchTasksByBatch', reason, { batchId }, { affected: count })
|
|
|
- }
|
|
|
- return count
|
|
|
- }
|
|
|
-
|
|
|
- serializeBatch(row) {
|
|
|
- return {
|
|
|
- id: row.id,
|
|
|
- name: row.name,
|
|
|
- jx0502zbid: row.jx0502zbid,
|
|
|
- enabled: Number(row.enabled) === 1,
|
|
|
- create_time: row.create_time,
|
|
|
- update_time: row.update_time
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- async listBatches(options = {}) {
|
|
|
- const enabledOnly = !!options.enabledOnly
|
|
|
- const where = enabledOnly ? 'WHERE enabled = 1' : ''
|
|
|
- const rows = await db.query(`SELECT * FROM qk_batch ${where} ORDER BY update_time DESC, id DESC`)
|
|
|
- return (rows || []).map(row => this.serializeBatch(row))
|
|
|
- }
|
|
|
-
|
|
|
- async createBatch(payload = {}) {
|
|
|
- const name = String(payload.name || '').trim()
|
|
|
- const jx0502zbid = String(payload.jx0502zbid || '').trim()
|
|
|
- if (!name || !jx0502zbid) {
|
|
|
- throw new Error('批次名称和批次 ID 不能为空')
|
|
|
- }
|
|
|
- const now = Date.now()
|
|
|
- const result = await db.query(
|
|
|
- 'INSERT INTO qk_batch (name, jx0502zbid, enabled, create_time, update_time) VALUES (?, ?, ?, ?, ?)',
|
|
|
- [name, jx0502zbid, payload.enabled === false ? 0 : 1, now, now]
|
|
|
- )
|
|
|
- if (!result || result.affectedRows <= 0) {
|
|
|
- throw new Error('创建抢课批次失败')
|
|
|
- }
|
|
|
- this.logInfo('createBatch', '抢课批次已创建', { batchId: result.insertId }, { name, jx0502zbid })
|
|
|
- return result.insertId
|
|
|
- }
|
|
|
-
|
|
|
- async updateBatch(batchId, payload = {}) {
|
|
|
- const existing = await this.getBatchById(batchId)
|
|
|
- const name = payload.name !== undefined ? String(payload.name || '').trim() : existing.name
|
|
|
- const jx0502zbid = payload.jx0502zbid !== undefined ? String(payload.jx0502zbid || '').trim() : existing.jx0502zbid
|
|
|
- const enabled = payload.enabled !== undefined ? (payload.enabled ? 1 : 0) : Number(existing.enabled)
|
|
|
- if (!name || !jx0502zbid) {
|
|
|
- throw new Error('批次名称和批次 ID 不能为空')
|
|
|
- }
|
|
|
- const now = Date.now()
|
|
|
- const result = await db.query(
|
|
|
- 'UPDATE qk_batch SET name = ?, jx0502zbid = ?, enabled = ?, update_time = ? WHERE id = ?',
|
|
|
- [name, jx0502zbid, enabled, now, batchId]
|
|
|
- )
|
|
|
- if (!result || result.affectedRows <= 0) {
|
|
|
- throw new Error('更新抢课批次失败')
|
|
|
- }
|
|
|
- const disabledNow = Number(existing.enabled) === 1 && enabled === 0
|
|
|
- const idChanged = String(existing.jx0502zbid) !== jx0502zbid
|
|
|
- if (disabledNow) {
|
|
|
- await this.pauseTasksByBatch(batchId)
|
|
|
- } else if (idChanged && enabled === 1) {
|
|
|
- await this.redispatchTasksByBatch(batchId)
|
|
|
- }
|
|
|
- this.logInfo('updateBatch', '抢课批次已更新', { batchId }, {
|
|
|
- name,
|
|
|
- jx0502zbid,
|
|
|
- enabled: enabled === 1,
|
|
|
- disabledNow,
|
|
|
- idChanged
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
- 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 })
|
|
|
- }
|
|
|
-
|
|
|
- serializeReportLog(row) {
|
|
|
- const result = { ...row }
|
|
|
- if (typeof result.payload_json === 'string' && result.payload_json) {
|
|
|
- try {
|
|
|
- result.payload_json = JSON.parse(result.payload_json)
|
|
|
- } catch (_) {}
|
|
|
- }
|
|
|
- const payload = result.payload_json && typeof result.payload_json === 'object'
|
|
|
- ? result.payload_json
|
|
|
- : {}
|
|
|
- result.course_name = this.extractReportCourseName(payload)
|
|
|
- result.display_message = this.formatReportMessage(result.message, payload)
|
|
|
- return result
|
|
|
- }
|
|
|
-
|
|
|
- buildReportMessage(payload = {}) {
|
|
|
- const base = (() => {
|
|
|
- if (payload.message) {
|
|
|
- return String(payload.message)
|
|
|
- }
|
|
|
- if (payload.error_msg) {
|
|
|
- return String(payload.error_msg)
|
|
|
- }
|
|
|
- if (payload.error) {
|
|
|
- return String(payload.error)
|
|
|
- }
|
|
|
- if (payload.success === true) {
|
|
|
- return payload.label ? `${payload.label} 成功` : '请求成功'
|
|
|
- }
|
|
|
- return payload.label ? `${payload.label} 失败` : '请求失败'
|
|
|
- })()
|
|
|
- return this.formatReportMessage(base, payload)
|
|
|
- }
|
|
|
-
|
|
|
- async assertClientTaskAccess(clientId, 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.assigned_client_id !== clientId) {
|
|
|
- throw new Error('任务不属于当前客户端')
|
|
|
- }
|
|
|
- if (![TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING].includes(task.status)) {
|
|
|
- throw new Error('任务当前状态不可上报')
|
|
|
- }
|
|
|
- return task
|
|
|
- }
|
|
|
-
|
|
|
- async reportProgress(clientId, clientSecret, payload = {}) {
|
|
|
- const client = await this.authenticateClient(clientId, clientSecret)
|
|
|
- if (!client) {
|
|
|
- throw new Error('客户端凭证无效')
|
|
|
- }
|
|
|
- const taskId = Number(payload.task_id || payload.id)
|
|
|
- if (!taskId) {
|
|
|
- throw new Error('缺少任务 ID')
|
|
|
- }
|
|
|
- const event = String(payload.event || 'request_result')
|
|
|
- if (!REPORT_LOG_EVENTS.has(event) || event === 'grab_success' || event === 'grab_fail') {
|
|
|
- throw new Error('不支持的上报类型')
|
|
|
- }
|
|
|
- const task = await this.assertClientTaskAccess(clientId, taskId)
|
|
|
- const message = this.buildReportMessage(payload)
|
|
|
- const now = Date.now()
|
|
|
- await this.logTask(taskId, clientId, event, message, payload)
|
|
|
- const updates = ['update_time = ?']
|
|
|
- const params = [now]
|
|
|
- if (task.status === TASK_STATUS.ASSIGNED) {
|
|
|
- updates.push('status = ?')
|
|
|
- params.push(TASK_STATUS.RUNNING)
|
|
|
- }
|
|
|
- if (payload.success !== true && message) {
|
|
|
- updates.push('error_msg = ?')
|
|
|
- params.push(message)
|
|
|
- }
|
|
|
- params.push(taskId, clientId)
|
|
|
- await db.query(
|
|
|
- `UPDATE qk_task SET ${updates.join(', ')} WHERE id = ? AND assigned_client_id = ?`,
|
|
|
- params
|
|
|
- )
|
|
|
- this.logInfo('reportProgress', '客户端上报抢课进度', { taskId, clientId }, {
|
|
|
- event,
|
|
|
- success: payload.success === true,
|
|
|
- message
|
|
|
- })
|
|
|
- return { task_id: taskId, event, message }
|
|
|
- }
|
|
|
-
|
|
|
- async createTask(uuid, payload) {
|
|
|
- await this.ensureSchema()
|
|
|
- const validated = this.validateTaskCoursesAndInterval(
|
|
|
- payload.courses || payload.COURSES,
|
|
|
- payload.course_groups || payload.COURSE_GROUPS,
|
|
|
- payload.interval_ms || payload.INTERVAL_MS || 500
|
|
|
- )
|
|
|
- const { courses, courseGroups, intervalMs } = validated
|
|
|
- const batchId = this.resolveTaskBatchId(payload)
|
|
|
- await this.requireEnabledBatch(batchId)
|
|
|
- const account = await this.resolveTaskAccount(uuid, payload, { requireAccount: true })
|
|
|
- const time = Date.now()
|
|
|
- const sql = `INSERT INTO qk_task
|
|
|
- (create_user, name, batch_id, jx0502zbid, student_num, jw_account_id, password_enc, courses, course_groups, enable_ggxxk, auto_relogin, interval_ms, status, create_time, update_time)
|
|
|
- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
|
|
|
- const result = await db.query(sql, [
|
|
|
- uuid,
|
|
|
- payload.name,
|
|
|
- batchId,
|
|
|
- '',
|
|
|
- account.student_num,
|
|
|
- account.jw_account_id,
|
|
|
- this.encryptPassword(account.password),
|
|
|
- JSON.stringify(courses),
|
|
|
- JSON.stringify(courseGroups),
|
|
|
- payload.enable_ggxxk || payload.ENABLE_GGXXK ? 1 : 0,
|
|
|
- this.parseAutoRelogin(payload) ? 1 : 0,
|
|
|
- intervalMs,
|
|
|
- TASK_STATUS.PAUSED,
|
|
|
- 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,
|
|
|
- batch_id: batchId,
|
|
|
- student_num: account.student_num,
|
|
|
- jw_account_id: account.jw_account_id,
|
|
|
- courses_count: courses.length,
|
|
|
- course_groups_count: courseGroups.length,
|
|
|
- interval_ms: intervalMs,
|
|
|
- enable_ggxxk: !!(payload.enable_ggxxk || payload.ENABLE_GGXXK),
|
|
|
- auto_relogin: this.parseAutoRelogin(payload)
|
|
|
- })
|
|
|
- return result.insertId
|
|
|
- }
|
|
|
-
|
|
|
- async updateTask(uuid, taskId, payload) {
|
|
|
- await this.ensureSchema()
|
|
|
- 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.PAUSED, TASK_STATUS.FAILED, TASK_STATUS.CANCELLED].includes(rows[0].status)) {
|
|
|
- throw new Error('任务进行中或已分配,请先暂停后再修改')
|
|
|
- }
|
|
|
- const validated = this.validateTaskCoursesAndInterval(
|
|
|
- payload.courses || payload.COURSES,
|
|
|
- payload.course_groups || payload.COURSE_GROUPS,
|
|
|
- payload.interval_ms || payload.INTERVAL_MS || 500
|
|
|
- )
|
|
|
- const { courses, courseGroups, intervalMs } = validated
|
|
|
- const batchId = this.resolveTaskBatchId(payload)
|
|
|
- await this.requireEnabledBatch(batchId)
|
|
|
- const account = await this.resolveTaskAccount(uuid, payload, { requireAccount: true })
|
|
|
- const params = [
|
|
|
- payload.name,
|
|
|
- batchId,
|
|
|
- account.student_num,
|
|
|
- account.jw_account_id,
|
|
|
- this.encryptPassword(account.password),
|
|
|
- JSON.stringify(courses),
|
|
|
- JSON.stringify(courseGroups),
|
|
|
- payload.enable_ggxxk || payload.ENABLE_GGXXK ? 1 : 0,
|
|
|
- this.parseAutoRelogin(payload) ? 1 : 0,
|
|
|
- intervalMs,
|
|
|
- TASK_STATUS.PAUSED,
|
|
|
- Date.now()
|
|
|
- ]
|
|
|
- params.push(taskId, uuid)
|
|
|
- const sql = `UPDATE qk_task SET name = ?, batch_id = ?, student_num = ?, jw_account_id = ?, password_enc = ?, courses = ?, course_groups = ?, enable_ggxxk = ?, auto_relogin = ?, interval_ms = ?, status = ?, jx0502zbid = '', 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: account.student_num,
|
|
|
- jw_account_id: account.jw_account_id,
|
|
|
- courses_count: courses.length,
|
|
|
- course_groups_count: courseGroups.length,
|
|
|
- interval_ms: intervalMs,
|
|
|
- account_changed: true
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
- async listUserTasks(uuid, filters = {}) {
|
|
|
- const where = ['t.create_user = ?']
|
|
|
- const params = [uuid]
|
|
|
- if (filters.status) {
|
|
|
- where.push('t.status = ?')
|
|
|
- params.push(filters.status)
|
|
|
- }
|
|
|
- if (filters.student_num) {
|
|
|
- where.push('t.student_num LIKE ?')
|
|
|
- params.push(`%${filters.student_num}%`)
|
|
|
- }
|
|
|
- if (filters.name) {
|
|
|
- where.push('t.name LIKE ?')
|
|
|
- params.push(`%${filters.name}%`)
|
|
|
- }
|
|
|
- const courseName = String(filters.course_name || filters.course || '').trim()
|
|
|
- if (courseName) {
|
|
|
- const keyword = `%${courseName}%`
|
|
|
- where.push('(t.courses LIKE ? OR t.course_groups LIKE ?)')
|
|
|
- params.push(keyword, keyword)
|
|
|
- }
|
|
|
- if (filters.batch_id) {
|
|
|
- where.push('t.batch_id = ?')
|
|
|
- params.push(Number(filters.batch_id))
|
|
|
- }
|
|
|
- const whereSql = where.join(' AND ')
|
|
|
- const rows = await db.query(
|
|
|
- `SELECT t.*${this.taskBatchSelectSql('t')}
|
|
|
- FROM qk_task t
|
|
|
- ${this.taskBatchJoinSql('t')}
|
|
|
- WHERE ${whereSql}
|
|
|
- ORDER BY t.create_time DESC`,
|
|
|
- params
|
|
|
- )
|
|
|
- return (rows || []).map(row => this.serializeTask(row))
|
|
|
- }
|
|
|
-
|
|
|
- async getTaskDetail(uuid, taskId) {
|
|
|
- const rows = await db.query(
|
|
|
- `SELECT t.*${this.taskBatchSelectSql('t')}
|
|
|
- FROM qk_task t
|
|
|
- ${this.taskBatchJoinSql('t')}
|
|
|
- WHERE t.id = ? AND t.create_user = ?`,
|
|
|
- [taskId, uuid]
|
|
|
- )
|
|
|
- if (!rows || rows.length === 0) {
|
|
|
- return null
|
|
|
- }
|
|
|
- const task = this.serializeTask(rows[0], false)
|
|
|
- task.jw_account_id = await this.inferJwAccountId(uuid, task.student_num, task.jw_account_id)
|
|
|
- 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,
|
|
|
- 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.PAUSED, 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 startTask(uuid, taskId) {
|
|
|
- const rows = await db.query('SELECT * FROM qk_task WHERE id = ? AND create_user = ?', [taskId, uuid])
|
|
|
- if (!rows || rows.length === 0) {
|
|
|
- throw new Error('任务不存在')
|
|
|
- }
|
|
|
- const task = rows[0]
|
|
|
- if (![TASK_STATUS.PAUSED, TASK_STATUS.FAILED].includes(task.status)) {
|
|
|
- throw new Error('仅未开始或失败的任务可开启')
|
|
|
- }
|
|
|
- if (task.batch_id) {
|
|
|
- await this.requireEnabledBatch(task.batch_id)
|
|
|
- }
|
|
|
- const now = Date.now()
|
|
|
- const result = await db.query(
|
|
|
- `UPDATE qk_task SET 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 = ? AND create_user = ? AND status IN (?, ?)`,
|
|
|
- [TASK_STATUS.PENDING, now, taskId, uuid, TASK_STATUS.PAUSED, TASK_STATUS.FAILED]
|
|
|
- )
|
|
|
- if (!result || result.affectedRows <= 0) {
|
|
|
- throw new Error('开启抢课任务失败')
|
|
|
- }
|
|
|
- await this.logTask(taskId, null, 'started', '用户开启抢课任务,等待分配')
|
|
|
- this.logInfo('startTask', '用户已开启抢课任务', { taskId, uuid })
|
|
|
- }
|
|
|
-
|
|
|
- async pauseTask(uuid, taskId) {
|
|
|
- const rows = await db.query('SELECT * FROM qk_task WHERE id = ? AND create_user = ?', [taskId, uuid])
|
|
|
- if (!rows || rows.length === 0) {
|
|
|
- throw new Error('任务不存在')
|
|
|
- }
|
|
|
- const task = rows[0]
|
|
|
- if (![TASK_STATUS.PENDING, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING].includes(task.status)) {
|
|
|
- throw new Error('当前状态不可暂停')
|
|
|
- }
|
|
|
- const now = Date.now()
|
|
|
- const wasAssigned = [TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING].includes(task.status)
|
|
|
- const releasedClientId = wasAssigned ? task.assigned_client_id : null
|
|
|
- const result = await db.query(
|
|
|
- `UPDATE qk_task SET status = ?, update_time = ?, assigned_client_id = NULL, lease_expire_at = NULL,
|
|
|
- assigned_at = NULL, exclude_client_id = NULL
|
|
|
- WHERE id = ? AND create_user = ? AND status IN (?, ?, ?)`,
|
|
|
- [TASK_STATUS.PAUSED, now, taskId, uuid, TASK_STATUS.PENDING, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
|
|
|
- )
|
|
|
- if (!result || result.affectedRows <= 0) {
|
|
|
- throw new Error('暂停抢课任务失败')
|
|
|
- }
|
|
|
- if (releasedClientId) {
|
|
|
- await this.syncClientSlots(releasedClientId, now)
|
|
|
- }
|
|
|
- await this.logTask(taskId, task.assigned_client_id, 'paused', wasAssigned ? '用户暂停任务,已收回客户端' : '用户暂停抢课任务')
|
|
|
- this.logInfo('pauseTask', '用户已暂停抢课任务', { taskId, uuid }, {
|
|
|
- released_from_client: wasAssigned ? task.assigned_client_id : null
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
- async getAdminTask(taskId) {
|
|
|
- const rows = await db.query(
|
|
|
- `SELECT t.*, u.username, u.avatar${this.taskBatchSelectSql('t')}
|
|
|
- FROM qk_task t
|
|
|
- LEFT JOIN users u ON u.uuid COLLATE utf8mb4_general_ci = t.create_user COLLATE utf8mb4_general_ci
|
|
|
- ${this.taskBatchJoinSql('t')}
|
|
|
- WHERE t.id = ?`,
|
|
|
- [taskId]
|
|
|
- )
|
|
|
- if (!rows || rows.length === 0) {
|
|
|
- return null
|
|
|
- }
|
|
|
- const task = this.serializeTask(rows[0], false)
|
|
|
- task.jw_account_id = await this.inferJwAccountId(rows[0].create_user, task.student_num, task.jw_account_id)
|
|
|
- return task
|
|
|
- }
|
|
|
-
|
|
|
- async adminUpdateTask(taskId, payload) {
|
|
|
- await this.ensureSchema()
|
|
|
- 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 validated = this.validateTaskCoursesAndInterval(
|
|
|
- payload.courses || payload.COURSES,
|
|
|
- payload.course_groups || payload.COURSE_GROUPS,
|
|
|
- payload.interval_ms || payload.INTERVAL_MS || task.interval_ms || 500
|
|
|
- )
|
|
|
- const { courses, courseGroups, intervalMs } = validated
|
|
|
- const now = Date.now()
|
|
|
- const wasAssigned = [TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING].includes(task.status)
|
|
|
- const releasedClientId = wasAssigned ? task.assigned_client_id : null
|
|
|
- const batchId = this.resolveTaskBatchId(payload)
|
|
|
- const account = await this.resolveTaskAccount(task.create_user, payload, { requireAccount: true })
|
|
|
- const params = [
|
|
|
- payload.name,
|
|
|
- batchId,
|
|
|
- account.student_num,
|
|
|
- account.jw_account_id,
|
|
|
- this.encryptPassword(account.password),
|
|
|
- JSON.stringify(courses),
|
|
|
- JSON.stringify(courseGroups),
|
|
|
- payload.enable_ggxxk || payload.ENABLE_GGXXK ? 1 : 0,
|
|
|
- this.parseAutoRelogin(payload) ? 1 : 0,
|
|
|
- intervalMs,
|
|
|
- TASK_STATUS.PAUSED,
|
|
|
- now,
|
|
|
- taskId
|
|
|
- ]
|
|
|
- const sql = `UPDATE qk_task SET name = ?, batch_id = ?, student_num = ?, jw_account_id = ?, password_enc = ?, courses = ?, course_groups = ?, enable_ggxxk = ?, auto_relogin = ?, interval_ms = ?, status = ?, jx0502zbid = '', 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('更新抢课任务失败')
|
|
|
- }
|
|
|
- if (releasedClientId) {
|
|
|
- await this.syncClientSlots(releasedClientId, now)
|
|
|
- }
|
|
|
- await this.logTask(taskId, task.assigned_client_id, 'admin_updated', wasAssigned ? '管理员更新任务并收回(未开始)' : '管理员更新抢课任务')
|
|
|
- this.logInfo('adminUpdateTask', '管理员已更新抢课任务', { taskId }, {
|
|
|
- name: payload.name,
|
|
|
- student_num: account.student_num,
|
|
|
- jw_account_id: account.jw_account_id,
|
|
|
- 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.PAUSED, TASK_STATUS.PENDING, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING, TASK_STATUS.FAILED].includes(task.status)) {
|
|
|
- throw new Error('当前状态不可取消')
|
|
|
- }
|
|
|
- const now = Date.now()
|
|
|
- const releasedClientId = [TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING].includes(task.status)
|
|
|
- ? task.assigned_client_id
|
|
|
- : null
|
|
|
- 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.PAUSED, TASK_STATUS.PENDING, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING, TASK_STATUS.FAILED]
|
|
|
- )
|
|
|
- if (!result || result.affectedRows <= 0) {
|
|
|
- throw new Error('取消抢课任务失败')
|
|
|
- }
|
|
|
- if (releasedClientId) {
|
|
|
- await this.syncClientSlots(releasedClientId, now)
|
|
|
- }
|
|
|
- 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 adminStartTask(taskId) {
|
|
|
- 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.PAUSED, TASK_STATUS.FAILED].includes(task.status)) {
|
|
|
- throw new Error('仅未开始或失败的任务可开启')
|
|
|
- }
|
|
|
- if (task.batch_id) {
|
|
|
- await this.requireEnabledBatch(task.batch_id)
|
|
|
- }
|
|
|
- const now = Date.now()
|
|
|
- const result = await db.query(
|
|
|
- `UPDATE qk_task SET 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 = ? AND status IN (?, ?)`,
|
|
|
- [TASK_STATUS.PENDING, now, taskId, TASK_STATUS.PAUSED, TASK_STATUS.FAILED]
|
|
|
- )
|
|
|
- if (!result || result.affectedRows <= 0) {
|
|
|
- throw new Error('开启抢课任务失败')
|
|
|
- }
|
|
|
- await this.logTask(taskId, null, 'admin_started', '管理员开启抢课任务,等待分配')
|
|
|
- this.logInfo('adminStartTask', '管理员已开启抢课任务', { taskId })
|
|
|
- }
|
|
|
-
|
|
|
- async adminPauseTask(taskId) {
|
|
|
- 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.PENDING, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING].includes(task.status)) {
|
|
|
- throw new Error('当前状态不可暂停')
|
|
|
- }
|
|
|
- const now = Date.now()
|
|
|
- const wasAssigned = [TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING].includes(task.status)
|
|
|
- const releasedClientId = wasAssigned ? task.assigned_client_id : null
|
|
|
- const result = await db.query(
|
|
|
- `UPDATE qk_task SET status = ?, update_time = ?, assigned_client_id = NULL, lease_expire_at = NULL,
|
|
|
- assigned_at = NULL, exclude_client_id = NULL
|
|
|
- WHERE id = ? AND status IN (?, ?, ?)`,
|
|
|
- [TASK_STATUS.PAUSED, now, taskId, TASK_STATUS.PENDING, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
|
|
|
- )
|
|
|
- if (!result || result.affectedRows <= 0) {
|
|
|
- throw new Error('暂停抢课任务失败')
|
|
|
- }
|
|
|
- if (releasedClientId) {
|
|
|
- await this.syncClientSlots(releasedClientId, now)
|
|
|
- }
|
|
|
- await this.logTask(taskId, task.assigned_client_id, 'admin_paused', wasAssigned ? '管理员暂停任务,已收回客户端' : '管理员暂停抢课任务')
|
|
|
- this.logInfo('adminPauseTask', '管理员已暂停抢课任务', { taskId }, {
|
|
|
- released_from_client: wasAssigned ? task.assigned_client_id : null
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
- 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 time = Date.now()
|
|
|
- await db.query(
|
|
|
- 'UPDATE qk_client SET label = COALESCE(NULLIF(?, \'\'), label), max_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,
|
|
|
- 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 })
|
|
|
- const currentSlots = await this.syncClientSlots(clientId, time)
|
|
|
- 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 revokedTaskIds = await this.findRevokedClientTasks(clientId, runningTasks)
|
|
|
- const activeRunningTasks = runningTasks.filter(id => !revokedTaskIds.includes(Number(id)))
|
|
|
- const maxSlots = this.resolveClientMaxSlots(client, payload)
|
|
|
- const now = Date.now()
|
|
|
- await db.query(
|
|
|
- 'UPDATE qk_client SET max_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,
|
|
|
- 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 })
|
|
|
- if (activeRunningTasks.length > 0) {
|
|
|
- const leaseExpireAt = now + this.leaseMs
|
|
|
- const placeholders = activeRunningTasks.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, ...activeRunningTasks, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
|
|
|
- )
|
|
|
- this.logInfo('heartbeat', '客户端心跳续租运行中任务', { clientId }, {
|
|
|
- running_tasks: activeRunningTasks,
|
|
|
- revoked_task_ids: revokedTaskIds,
|
|
|
- lease_expire_at: leaseExpireAt,
|
|
|
- max_slots: maxSlots
|
|
|
- })
|
|
|
- } else if (revokedTaskIds.length > 0) {
|
|
|
- this.logInfo('heartbeat', '客户端上报任务已被服务端收回', { clientId }, {
|
|
|
- revoked_task_ids: revokedTaskIds
|
|
|
- })
|
|
|
- }
|
|
|
- await this.releaseOrphanedClientTasks(clientId, activeRunningTasks, now)
|
|
|
- const currentSlots = await this.syncClientSlots(clientId, now)
|
|
|
- return { client_id: clientId, current_slots: currentSlots, max_slots: maxSlots, revoked_task_ids: revokedTaskIds }
|
|
|
- }
|
|
|
-
|
|
|
- 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 })
|
|
|
- }
|
|
|
- }
|
|
|
- if (released > 0) {
|
|
|
- await this.syncClientSlots(clientId, now)
|
|
|
- }
|
|
|
- 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 t.*, b.jx0502zbid AS batch_jx0502zbid, b.name AS batch_name, b.enabled AS batch_enabled
|
|
|
- FROM qk_task t
|
|
|
- LEFT JOIN qk_batch b ON b.id = t.batch_id
|
|
|
- WHERE t.assigned_client_id = ? AND t.status IN (?, ?)
|
|
|
- ORDER BY t.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) {
|
|
|
- await this.syncClientSlots(clientId, now)
|
|
|
- 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 = await this.getClientAvailableSlotsAsync(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 inFlightStatuses = this.getInFlightTaskStatuses()
|
|
|
- const candidateLimit = Math.min(50, Math.max(safeCount, safeCount * 5))
|
|
|
- const [candidateRows] = await conn.execute(
|
|
|
- `SELECT t.*, b.jx0502zbid AS batch_jx0502zbid, b.name AS batch_name, b.enabled AS batch_enabled
|
|
|
- FROM qk_task t
|
|
|
- LEFT JOIN qk_batch b ON b.id = t.batch_id
|
|
|
- WHERE t.status = ?
|
|
|
- AND (t.batch_id IS NULL OR b.enabled = 1)
|
|
|
- AND (t.exclude_client_id IS NULL OR t.exclude_client_id <> ?)
|
|
|
- AND NOT EXISTS (
|
|
|
- SELECT 1 FROM qk_task active
|
|
|
- WHERE active.student_num = t.student_num
|
|
|
- AND active.status IN (?, ?)
|
|
|
- )
|
|
|
- ORDER BY t.create_time ASC LIMIT ${candidateLimit} FOR UPDATE`,
|
|
|
- [TASK_STATUS.PENDING, clientId, ...inFlightStatuses]
|
|
|
- )
|
|
|
- const rows = this.pickPullableTasks(candidateRows, safeCount)
|
|
|
- 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 this.syncClientSlots(clientId, now)
|
|
|
- }
|
|
|
- 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)
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- extractFailureMessage(payload = {}) {
|
|
|
- const result = payload.result && typeof payload.result === 'object' ? payload.result : {}
|
|
|
- const parts = [
|
|
|
- payload.message,
|
|
|
- payload.error_msg,
|
|
|
- payload.error,
|
|
|
- result.message,
|
|
|
- result.error,
|
|
|
- result.error_msg,
|
|
|
- typeof payload.result === 'string' ? payload.result : null
|
|
|
- ]
|
|
|
- return parts.filter(Boolean).map(item => String(item)).join(' ').trim()
|
|
|
- }
|
|
|
-
|
|
|
- resolveFailureType(payload = {}) {
|
|
|
- const result = payload.result && typeof payload.result === 'object' ? payload.result : {}
|
|
|
- return String(payload.type || payload.event || result.type || result.event || '').trim()
|
|
|
- }
|
|
|
-
|
|
|
- shouldRequeueTaskOnFailure(task, payload = {}) {
|
|
|
- if (payload.success === true) {
|
|
|
- return false
|
|
|
- }
|
|
|
- if (payload.requeue === true || payload.requeue === 1 || payload.requeue === '1') {
|
|
|
- return true
|
|
|
- }
|
|
|
-
|
|
|
- const message = this.extractFailureMessage(payload)
|
|
|
- const failureType = this.resolveFailureType(payload)
|
|
|
- const terminalTypes = new Set(['already', 'dajia', 'not_open', 'not_in_time'])
|
|
|
- if (terminalTypes.has(failureType)) {
|
|
|
- return false
|
|
|
- }
|
|
|
-
|
|
|
- const terminalPatterns = [
|
|
|
- '已选择',
|
|
|
- '冲突',
|
|
|
- '超过',
|
|
|
- '选课不开放',
|
|
|
- '不在选课时间',
|
|
|
- '用户名或密码',
|
|
|
- '未匹配到目标课程',
|
|
|
- '验证码验证失败次数',
|
|
|
- '为避免账号被锁定'
|
|
|
- ]
|
|
|
- if (terminalPatterns.some(pattern => message.includes(pattern))) {
|
|
|
- return false
|
|
|
- }
|
|
|
-
|
|
|
- // 旧版客户端 task_exit 上报:抢课任务异常结束,code=1
|
|
|
- if (/抢课任务异常结束/.test(message) || /\bcode=1\b/.test(message)) {
|
|
|
- return true
|
|
|
- }
|
|
|
- if (/抢课循环结束但未获得成功结果/.test(message)) {
|
|
|
- return true
|
|
|
- }
|
|
|
-
|
|
|
- // 客户端被收回槽位、窗口关闭等导致的被动停止,应重新分配而非标为失败
|
|
|
- const exitReason = String(payload.exit_reason || payload.reason || '').trim()
|
|
|
- if (failureType === 'stopped' || exitReason === 'stopped' || message.includes('任务已停止')) {
|
|
|
- return true
|
|
|
- }
|
|
|
-
|
|
|
- const autoRelogin = Number(task?.auto_relogin) === 1
|
|
|
- if (autoRelogin) {
|
|
|
- // 新版可传 type=login_expired;旧版仅 message / result.message
|
|
|
- if (failureType === 'login_expired') {
|
|
|
- return true
|
|
|
- }
|
|
|
- const loginPatterns = [
|
|
|
- '别处登录',
|
|
|
- '在其他地方登录',
|
|
|
- '账号在别处',
|
|
|
- '登录失效',
|
|
|
- '登录已失效',
|
|
|
- '请重新登录',
|
|
|
- '按登录失效处理',
|
|
|
- '空响应',
|
|
|
- '未登录',
|
|
|
- '登录失败',
|
|
|
- '会话过期',
|
|
|
- '会话失效'
|
|
|
- ]
|
|
|
- if (loginPatterns.some(pattern => message.includes(pattern))) {
|
|
|
- return true
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return false
|
|
|
- }
|
|
|
-
|
|
|
- async requeueTaskAfterClientFailure(clientId, taskId, payload = {}) {
|
|
|
- const now = Date.now()
|
|
|
- const message = this.buildReportMessage(payload) || '任务异常结束,已重新进入待分配队列'
|
|
|
- const result = await db.query(
|
|
|
- `UPDATE qk_task SET status = ?, assigned_client_id = NULL, lease_expire_at = NULL,
|
|
|
- assigned_at = NULL, result_json = NULL, error_msg = ?, finished_time = NULL, update_time = ?
|
|
|
- WHERE id = ? AND assigned_client_id = ? AND status IN (?, ?)`,
|
|
|
- [TASK_STATUS.PENDING, message, now, taskId, clientId, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
|
|
|
- )
|
|
|
- if (!result || result.affectedRows <= 0) {
|
|
|
- throw new Error('任务不存在或不属于当前客户端')
|
|
|
- }
|
|
|
- await this.syncClientSlots(clientId, now)
|
|
|
- await this.logTask(taskId, clientId, 'requeued', message, payload.result || payload)
|
|
|
- this.logInfo('reportResult', '抢课任务异常结束,已重新入队', { taskId, clientId }, {
|
|
|
- message,
|
|
|
- requeued: true,
|
|
|
- exit_reason: payload.exit_reason || payload.reason || null
|
|
|
- })
|
|
|
- return { task_id: taskId, status: TASK_STATUS.PENDING, requeued: true }
|
|
|
- }
|
|
|
-
|
|
|
- async reportResult(clientId, clientSecret, payload = {}) {
|
|
|
- await this.ensureSchema()
|
|
|
- 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 taskRows = await db.query(
|
|
|
- 'SELECT id, auto_relogin FROM qk_task WHERE id = ? AND assigned_client_id = ? AND status IN (?, ?)',
|
|
|
- [taskId, clientId, TASK_STATUS.ASSIGNED, TASK_STATUS.RUNNING]
|
|
|
- )
|
|
|
- if (!taskRows || taskRows.length === 0) {
|
|
|
- this.logWarn('reportResult', '任务结果上报被拒绝:任务不存在或不属于当前客户端', { taskId, clientId }, {
|
|
|
- success,
|
|
|
- status: success ? TASK_STATUS.SUCCESS : TASK_STATUS.FAILED
|
|
|
- })
|
|
|
- throw new Error('任务不存在或不属于当前客户端')
|
|
|
- }
|
|
|
- const task = taskRows[0]
|
|
|
-
|
|
|
- if (!success && this.shouldRequeueTaskOnFailure(task, payload)) {
|
|
|
- return this.requeueTaskAfterClientFailure(clientId, taskId, payload)
|
|
|
- }
|
|
|
-
|
|
|
- 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) {
|
|
|
- throw new Error('任务不存在或不属于当前客户端')
|
|
|
- }
|
|
|
- await this.syncClientSlots(clientId, now)
|
|
|
- await db.query(
|
|
|
- `UPDATE qk_client SET 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 enableClient(clientId) {
|
|
|
- const rows = await db.query('SELECT client_id, enabled FROM qk_client WHERE client_id = ?', [clientId])
|
|
|
- if (!rows || rows.length === 0) {
|
|
|
- throw new Error('客户端不存在')
|
|
|
- }
|
|
|
- if (Number(rows[0].enabled) === 1) {
|
|
|
- throw new Error('客户端已处于启用状态')
|
|
|
- }
|
|
|
- const now = Date.now()
|
|
|
- await db.query('UPDATE qk_client SET enabled = 1, update_time = ? WHERE client_id = ?', [now, clientId])
|
|
|
- this.logInfo('enableClient', '抢课客户端已解禁', { 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 courseName = String(filters.course_name || filters.course || '').trim()
|
|
|
- if (courseName) {
|
|
|
- const keyword = `%${courseName}%`
|
|
|
- where.push('(t.courses LIKE ? OR t.course_groups LIKE ?)')
|
|
|
- params.push(keyword, keyword)
|
|
|
- countParams.push(keyword, keyword)
|
|
|
- }
|
|
|
- if (filters.batch_id) {
|
|
|
- where.push('t.batch_id = ?')
|
|
|
- params.push(Number(filters.batch_id))
|
|
|
- countParams.push(Number(filters.batch_id))
|
|
|
- }
|
|
|
- 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${this.taskBatchSelectSql('t')}
|
|
|
- FROM qk_task t
|
|
|
- LEFT JOIN users u ON u.uuid COLLATE utf8mb4_general_ci = t.create_user COLLATE utf8mb4_general_ci
|
|
|
- ${this.taskBatchJoinSql('t')}
|
|
|
- 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 listAdminReports(filters = {}) {
|
|
|
- const pagesize = Math.max(1, Math.min(100, Number(filters.pagesize || 20)))
|
|
|
- const current = Math.max(1, Number(filters.current || 1))
|
|
|
- const where = ['l.event IN (?, ?, ?, ?)']
|
|
|
- const params = ['request_result', 'progress_snapshot', 'grab_success', 'grab_fail']
|
|
|
- const countParams = ['request_result', 'progress_snapshot', 'grab_success', 'grab_fail']
|
|
|
- if (filters.task_id) {
|
|
|
- where.push('l.task_id = ?')
|
|
|
- params.push(Number(filters.task_id))
|
|
|
- countParams.push(Number(filters.task_id))
|
|
|
- }
|
|
|
- if (filters.client_id) {
|
|
|
- where.push('l.client_id = ?')
|
|
|
- params.push(String(filters.client_id))
|
|
|
- countParams.push(String(filters.client_id))
|
|
|
- }
|
|
|
- if (filters.event) {
|
|
|
- where.push('l.event = ?')
|
|
|
- params.push(String(filters.event))
|
|
|
- countParams.push(String(filters.event))
|
|
|
- }
|
|
|
- if (filters.start_time) {
|
|
|
- where.push('l.create_time >= ?')
|
|
|
- params.push(Number(filters.start_time))
|
|
|
- countParams.push(Number(filters.start_time))
|
|
|
- }
|
|
|
- if (filters.end_time) {
|
|
|
- where.push('l.create_time <= ?')
|
|
|
- params.push(Number(filters.end_time))
|
|
|
- countParams.push(Number(filters.end_time))
|
|
|
- }
|
|
|
- if (filters.student_num) {
|
|
|
- where.push('t.student_num LIKE ?')
|
|
|
- params.push(`%${filters.student_num}%`)
|
|
|
- countParams.push(`%${filters.student_num}%`)
|
|
|
- }
|
|
|
- if (filters.name || filters.task_name) {
|
|
|
- where.push('t.name LIKE ?')
|
|
|
- params.push(`%${filters.name || filters.task_name}%`)
|
|
|
- countParams.push(`%${filters.name || filters.task_name}%`)
|
|
|
- }
|
|
|
- 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}%`)
|
|
|
- }
|
|
|
- if (filters.client_label) {
|
|
|
- where.push('c.label LIKE ?')
|
|
|
- params.push(`%${filters.client_label}%`)
|
|
|
- countParams.push(`%${filters.client_label}%`)
|
|
|
- }
|
|
|
- const whereSql = where.join(' AND ')
|
|
|
- const offset = (current - 1) * pagesize
|
|
|
- const joinSql = `
|
|
|
- FROM qk_task_log l
|
|
|
- LEFT JOIN qk_task t ON t.id = l.task_id
|
|
|
- LEFT JOIN qk_client c ON c.client_id = l.client_id
|
|
|
- LEFT JOIN users u ON u.uuid COLLATE utf8mb4_general_ci = t.create_user COLLATE utf8mb4_general_ci`
|
|
|
- const countRows = await db.query(
|
|
|
- `SELECT COUNT(*) AS total ${joinSql} WHERE ${whereSql}`,
|
|
|
- countParams
|
|
|
- )
|
|
|
- const rows = await db.query(
|
|
|
- `SELECT l.id, l.task_id, l.client_id, l.event, l.message, l.payload_json, l.create_time,
|
|
|
- t.name AS task_name, t.student_num, t.status AS task_status,
|
|
|
- c.label AS client_label, u.username, u.avatar
|
|
|
- ${joinSql}
|
|
|
- WHERE ${whereSql}
|
|
|
- ORDER BY l.create_time DESC
|
|
|
- LIMIT ${pagesize} OFFSET ${offset}`,
|
|
|
- params
|
|
|
- )
|
|
|
- return {
|
|
|
- list: (rows || []).map(row => this.serializeReportLog(row)),
|
|
|
- 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
|
|
|
- const affectedClients = new Set()
|
|
|
- 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) {
|
|
|
- affectedClients.add(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 })
|
|
|
- }
|
|
|
- }
|
|
|
- for (const clientId of affectedClients) {
|
|
|
- await this.syncClientSlots(clientId, now)
|
|
|
- }
|
|
|
- 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
|
|
|
- const affectedClients = new Set()
|
|
|
- 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) {
|
|
|
- affectedClients.add(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
|
|
|
- })
|
|
|
- }
|
|
|
- }
|
|
|
- for (const clientId of affectedClients) {
|
|
|
- await this.syncClientSlots(clientId, now)
|
|
|
- }
|
|
|
- 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
|
|
|
-}
|