'use strict'; const crypto = require('crypto'); const database = require('../config/database'); const TASK_TABLE = 'desktop_execution_tasks'; const EVENT_TABLE = 'desktop_execution_task_events'; const TASK_TYPES = new Set([ 'auto_listing', 'presale', 'review', 'campaign', 'promotion', 'delisting', 'product_test', ]); const ACTIVE_STATUSES = new Set(['queued', 'claimed', 'running']); const TERMINAL_STATUSES = new Set(['succeeded', 'failed', 'cancelled']); const DEFAULT_LEASE_SECONDS = 15 * 60; const MAX_LEASE_SECONDS = 60 * 60; let schemaReady = false; let schemaPromise = null; class DesktopExecutionTaskError extends Error { constructor(message, status = 400, code = 'DESKTOP_TASK_ERROR') { super(message); this.name = 'DesktopExecutionTaskError'; this.status = status; this.code = code; } } function normalizeText(value, maxLength = 255) { return String(value || '').trim().slice(0, maxLength); } function normalizePositiveInteger(value, fallback = 0) { const parsed = Number.parseInt(value, 10); return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; } function normalizeTaskType(value) { const type = normalizeText(value, 40).toLowerCase().replace(/-/g, '_'); if (!TASK_TYPES.has(type)) { throw new DesktopExecutionTaskError('不支持的桌面任务类型', 400, 'DESKTOP_TASK_TYPE_INVALID'); } return type; } function normalizeTaskStatus(value) { const status = normalizeText(value, 24).toLowerCase(); return [...ACTIVE_STATUSES, ...TERMINAL_STATUSES].includes(status) ? status : ''; } function normalizeJsonObject(value, fieldName = '任务参数') { if (value == null || value === '') return {}; if (typeof value === 'string') { try { const parsed = JSON.parse(value); if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) return parsed; } catch { // Use the validation error below. } throw new DesktopExecutionTaskError(`${fieldName}格式无效`, 400, 'DESKTOP_TASK_PAYLOAD_INVALID'); } if (typeof value !== 'object' || Array.isArray(value)) { throw new DesktopExecutionTaskError(`${fieldName}必须是对象`, 400, 'DESKTOP_TASK_PAYLOAD_INVALID'); } return value; } function parseStoredJson(value) { if (!value) return {}; if (typeof value === 'object') return value; try { const parsed = JSON.parse(value); return parsed && typeof parsed === 'object' ? parsed : {}; } catch { return {}; } } function toMysqlDateTime(value) { if (!value) return null; const date = value instanceof Date ? value : new Date(value); if (Number.isNaN(date.getTime())) return null; return date.toISOString().slice(0, 19).replace('T', ' '); } function buildTaskCode(now = new Date(), randomBytes = crypto.randomBytes) { const stamp = now.toISOString().replace(/[-:TZ.]/g, '').slice(0, 14); return `DT${stamp}${randomBytes(4).toString('hex').toUpperCase()}`; } function hashClaimToken(value) { return crypto.createHash('sha256').update(String(value || ''), 'utf8').digest('hex'); } function mapTask(row = {}) { return { id: Number(row.id || 0), taskCode: row.task_code || row.taskCode || '', taskType: row.task_type || row.taskType || '', title: row.title || '', assignedUserId: Number(row.assigned_user_id || row.assignedUserId || 0), assignedUserName: row.assigned_user_name || row.assignedUserName || '', assignedUsername: row.assigned_username || row.assignedUsername || '', targetShopId: row.target_shop_id || row.targetShopId || '', targetShopName: row.target_shop_name || row.targetShopName || '', sourceType: row.source_type || row.sourceType || '', sourceId: row.source_id || row.sourceId || '', sourceCode: row.source_code || row.sourceCode || '', status: row.status || 'queued', priority: Number(row.priority || 0), scheduledAt: row.scheduled_at || row.scheduledAt || null, expiresAt: row.expires_at || row.expiresAt || null, input: parseStoredJson(row.input_json ?? row.input), ruleSnapshot: parseStoredJson(row.rule_snapshot_json ?? row.ruleSnapshot), progress: { stage: row.progress_stage || row.progressStage || '', current: Number(row.progress_current || row.progressCurrent || 0), total: Number(row.progress_total || row.progressTotal || 0), percent: Number(row.progress_percent || row.progressPercent || 0), message: row.progress_message || row.progressMessage || '', }, claimedClientId: row.claimed_client_id || row.claimedClientId || '', claimExpiresAt: row.claim_expires_at || row.claimExpiresAt || null, attemptCount: Number(row.attempt_count || row.attemptCount || 0), result: parseStoredJson(row.result_json ?? row.result), errorMessage: row.error_message || row.errorMessage || '', createdByUserId: Number(row.created_by_user_id || row.createdByUserId || 0), createdByName: row.created_by_name || row.createdByName || '', startedAt: row.started_at || row.startedAt || null, completedAt: row.completed_at || row.completedAt || null, cancelledAt: row.cancelled_at || row.cancelledAt || null, createdAt: row.created_at || row.createdAt || null, updatedAt: row.updated_at || row.updatedAt || null, }; } async function ensureDesktopExecutionTaskSchema(db = database.local) { if (db === database.local && schemaReady) return true; if (db === database.local && schemaPromise) return schemaPromise; const ensure = async () => { await db.query(` CREATE TABLE IF NOT EXISTS ${TASK_TABLE} ( id BIGINT PRIMARY KEY AUTO_INCREMENT, task_code VARCHAR(40) NOT NULL, task_type VARCHAR(40) NOT NULL, title VARCHAR(255) NOT NULL, assigned_user_id BIGINT NOT NULL, assigned_user_name VARCHAR(120) NULL, assigned_username VARCHAR(120) NULL, target_shop_id VARCHAR(160) NULL, target_shop_name VARCHAR(255) NULL, source_type VARCHAR(40) NULL, source_id VARCHAR(160) NULL, source_code VARCHAR(160) NULL, status VARCHAR(24) NOT NULL DEFAULT 'queued', priority INT NOT NULL DEFAULT 0, scheduled_at DATETIME NULL, expires_at DATETIME NULL, input_json LONGTEXT NULL, rule_snapshot_json LONGTEXT NULL, progress_stage VARCHAR(80) NULL, progress_current INT NOT NULL DEFAULT 0, progress_total INT NOT NULL DEFAULT 0, progress_percent DECIMAL(6,2) NOT NULL DEFAULT 0, progress_message VARCHAR(500) NULL, claimed_client_id VARCHAR(64) NULL, claim_token_hash CHAR(64) NULL, claim_expires_at DATETIME NULL, attempt_count INT NOT NULL DEFAULT 0, result_json LONGTEXT NULL, error_message VARCHAR(1000) NULL, created_by_user_id BIGINT NOT NULL, created_by_name VARCHAR(120) NULL, started_at DATETIME NULL, completed_at DATETIME NULL, cancelled_at DATETIME NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_desktop_execution_task_code (task_code), UNIQUE KEY uk_desktop_execution_task_source (task_type, source_type, source_id), KEY idx_desktop_execution_assignee_status (assigned_user_id, status, scheduled_at, priority), KEY idx_desktop_execution_updated (updated_at), KEY idx_desktop_execution_client (claimed_client_id, status) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='总控下发的桌面执行任务' `); for (const statement of [ `ALTER TABLE ${TASK_TABLE} ADD COLUMN source_type VARCHAR(40) NULL AFTER target_shop_name`, `ALTER TABLE ${TASK_TABLE} ADD COLUMN source_id VARCHAR(160) NULL AFTER source_type`, `ALTER TABLE ${TASK_TABLE} ADD COLUMN source_code VARCHAR(160) NULL AFTER source_id`, `ALTER TABLE ${TASK_TABLE} ADD UNIQUE KEY uk_desktop_execution_task_source (task_type, source_type, source_id)`, ]) { try { await db.query(statement); } catch (error) { if (!['ER_DUP_FIELDNAME', 'ER_DUP_KEYNAME'].includes(error?.code)) throw error; } } await db.query(` CREATE TABLE IF NOT EXISTS ${EVENT_TABLE} ( id BIGINT PRIMARY KEY AUTO_INCREMENT, task_id BIGINT NOT NULL, task_code VARCHAR(40) NOT NULL, event_type VARCHAR(40) NOT NULL, from_status VARCHAR(24) NULL, to_status VARCHAR(24) NULL, actor_type VARCHAR(24) NOT NULL DEFAULT 'system', actor_id VARCHAR(120) NULL, actor_name VARCHAR(120) NULL, client_id VARCHAR(64) NULL, message VARCHAR(500) NULL, payload_json LONGTEXT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, KEY idx_desktop_execution_event_task (task_id, created_at), KEY idx_desktop_execution_event_code (task_code, created_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='桌面任务执行事件' `); if (db === database.local) schemaReady = true; return true; }; if (db !== database.local) return ensure(); schemaPromise = ensure().finally(() => { schemaPromise = null; }); return schemaPromise; } async function recordEvent(db, task, input = {}) { await db.insert( `INSERT INTO ${EVENT_TABLE} ( task_id, task_code, event_type, from_status, to_status, actor_type, actor_id, actor_name, client_id, message, payload_json, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())`, [ task.id, task.taskCode, normalizeText(input.eventType, 40), normalizeTaskStatus(input.fromStatus) || null, normalizeTaskStatus(input.toStatus) || null, normalizeText(input.actorType || 'system', 24), normalizeText(input.actorId, 120) || null, normalizeText(input.actorName, 120) || null, normalizeText(input.clientId, 64) || null, normalizeText(input.message, 500) || null, input.payload ? JSON.stringify(normalizeJsonObject(input.payload, '事件参数')) : null, ] ); } async function getTaskById(taskId, options = {}) { const db = options.db || database.local; await ensureDesktopExecutionTaskSchema(db); const rows = await db.query(`SELECT * FROM ${TASK_TABLE} WHERE id = ? LIMIT 1`, [Number(taskId)]); return rows[0] ? mapTask(rows[0]) : null; } async function createDesktopExecutionTask(input = {}, actor = {}, options = {}) { const db = options.db || database.local; await ensureDesktopExecutionTaskSchema(db); const taskType = normalizeTaskType(input.taskType || input.type); const assignedUserId = normalizePositiveInteger(input.assignedUserId || input.userId); if (!assignedUserId) { throw new DesktopExecutionTaskError('请选择任务执行人', 400, 'DESKTOP_TASK_ASSIGNEE_REQUIRED'); } const payload = normalizeJsonObject(input.input || input.payload, '任务参数'); const title = normalizeText(input.title, 255) || ({ presale: '自动预售处理', auto_listing: '自动上架', product_test: '商品测款', }[taskType] || '桌面执行任务'); const taskCode = buildTaskCode(options.now || new Date(), options.randomBytes || crypto.randomBytes); const priority = Math.max(-10, Math.min(10, Number.parseInt(input.priority, 10) || 0)); const scheduledAt = toMysqlDateTime(input.scheduledAt); const expiresAt = toMysqlDateTime(input.expiresAt); const result = await db.insert( `INSERT INTO ${TASK_TABLE} ( task_code, task_type, title, assigned_user_id, assigned_user_name, assigned_username, target_shop_id, target_shop_name, source_type, source_id, source_code, status, priority, scheduled_at, expires_at, input_json, rule_snapshot_json, created_by_user_id, created_by_name, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?, NOW(), NOW())`, [ taskCode, taskType, title, assignedUserId, normalizeText(input.assignedUserName, 120) || null, normalizeText(input.assignedUsername, 120) || null, normalizeText(input.targetShopId, 160) || null, normalizeText(input.targetShopName, 255) || null, normalizeText(input.sourceType, 40) || null, normalizeText(input.sourceId, 160) || null, normalizeText(input.sourceCode, 160) || null, priority, scheduledAt, expiresAt, JSON.stringify(payload), JSON.stringify(normalizeJsonObject(input.ruleSnapshot, '规则快照')), normalizePositiveInteger(actor.userId || actor.id), normalizeText(actor.name || actor.username, 120) || null, ] ); const task = await getTaskById(result.insertId, { db }); await recordEvent(db, task, { eventType: 'created', toStatus: 'queued', actorType: 'control_center', actorId: actor.userId || actor.id, actorName: actor.name || actor.username, message: '总控已下发任务', }); return task; } function buildAutoListingDesktopTaskInputs(plan = {}, actor = {}) { const listingShopSnapshot = plan.listingShopSnapshot && typeof plan.listingShopSnapshot === 'object' ? plan.listingShopSnapshot : {}; const assignedUserId = normalizePositiveInteger( listingShopSnapshot.executorUserId || actor.userId || actor.id ); const executorClientId = normalizeText( listingShopSnapshot.executorClientId || listingShopSnapshot.clientId, 64 ).toLowerCase(); const targetShopId = normalizeText(plan.listingShopId || plan.listingShop?.shopId, 160); const targetShopName = normalizeText(plan.listingShopName || plan.listingShop?.shopName, 255); if (!assignedUserId || !targetShopId) return []; return (Array.isArray(plan.listingTasks) ? plan.listingTasks : []) .filter((item) => ( item?.id && item?.selectedSpuCode && !item?.listedItemId && ['pending', 'failed', 'claimed'].includes(item?.listingStatus || 'pending') )) .map((item) => ({ taskType: 'auto_listing', title: `自动上架 · ${normalizeText(item.selectedProductTitle || item.selectedSpuName || item.taskCode, 220)}`, assignedUserId, assignedUserName: normalizeText( listingShopSnapshot.executorUserName || actor.name || plan.dingtalkUserName, 120 ), assignedUsername: normalizeText( listingShopSnapshot.executorUsername || actor.username || plan.dingtalkUsername, 120 ), targetShopId, targetShopName, sourceType: 'selection_fission_item', sourceId: String(item.id), sourceCode: normalizeText(item.taskCode || item.id, 160), input: { sourceTaskId: Number(item.id), sourceTaskCode: normalizeText(item.taskCode || item.id, 160), planId: Number(plan.id || 0), planCode: normalizeText(plan.schemeCode, 160), spuCode: normalizeText(item.selectedSpuCode, 160), productTitle: normalizeText(item.selectedProductTitle || item.selectedSpuName, 255), listingShopId: targetShopId, listingShopName: targetShopName, executorClientId, executorSessionId: normalizeText(listingShopSnapshot.sessionId, 64), routingMode: normalizeText(listingShopSnapshot.routingMode || 'self', 20), requestedByUserId: normalizePositiveInteger(actor.userId || actor.id), requestedByUserName: normalizeText(actor.name || actor.username, 120), }, ruleSnapshot: { executor: 'selection-fission-auto-listing-v1', authoritativeClaim: 'selection-fission', requireOnlineClient: true, requireAuthenticatedShop: true, executorClientId, }, })); } async function upsertDesktopExecutionTaskFromSource(input = {}, actor = {}, options = {}) { const db = options.db || database.local; await ensureDesktopExecutionTaskSchema(db); const taskType = normalizeTaskType(input.taskType || input.type); const sourceType = normalizeText(input.sourceType, 40); const sourceId = normalizeText(input.sourceId, 160); if (!sourceType || !sourceId) { throw new DesktopExecutionTaskError('来源任务身份不完整', 400, 'DESKTOP_TASK_SOURCE_REQUIRED'); } const findExisting = async () => { const rows = await db.query( `SELECT * FROM ${TASK_TABLE} WHERE task_type = ? AND source_type = ? AND source_id = ? LIMIT 1`, [taskType, sourceType, sourceId] ); return rows[0] ? mapTask(rows[0]) : null; }; let existing = await findExisting(); if (!existing) { try { return await createDesktopExecutionTask({ ...input, taskType, sourceType, sourceId }, actor, { ...options, db }); } catch (error) { if (error?.code !== 'ER_DUP_ENTRY') throw error; existing = await findExisting(); } } if (!existing) throw new DesktopExecutionTaskError('来源任务同步失败', 500, 'DESKTOP_TASK_SOURCE_SYNC_FAILED'); const reset = TERMINAL_STATUSES.has(existing.status); const updateAssignment = existing.status === 'queued' || reset; await db.update( `UPDATE ${TASK_TABLE} SET title = ?, source_code = ?, input_json = ?, rule_snapshot_json = ?, ${updateAssignment ? 'assigned_user_id = ?, assigned_user_name = ?, assigned_username = ?, target_shop_id = ?, target_shop_name = ?,' : ''} ${reset ? `status = 'queued', progress_stage = '', progress_current = 0, progress_total = 0, progress_percent = 0, progress_message = '等待桌面客户端领取', claimed_client_id = NULL, claim_token_hash = NULL, claim_expires_at = NULL, result_json = NULL, error_message = NULL, started_at = NULL, completed_at = NULL, cancelled_at = NULL,` : ''} updated_at = NOW() WHERE id = ?`, [ normalizeText(input.title, 255) || existing.title, normalizeText(input.sourceCode, 160) || null, JSON.stringify(normalizeJsonObject(input.input || input.payload, '任务参数')), JSON.stringify(normalizeJsonObject(input.ruleSnapshot, '规则快照')), ...(updateAssignment ? [ normalizePositiveInteger(input.assignedUserId || input.userId), normalizeText(input.assignedUserName, 120) || null, normalizeText(input.assignedUsername, 120) || null, normalizeText(input.targetShopId, 160) || null, normalizeText(input.targetShopName, 255) || null, ] : []), existing.id, ] ); const updated = await getTaskById(existing.id, { db }); if (reset) { await recordEvent(db, updated, { eventType: 'source_requeued', fromStatus: existing.status, toStatus: 'queued', actorType: 'control_center', actorId: actor.userId || actor.id, actorName: actor.name || actor.username, message: '来源上架任务重新进入执行队列', }); } return updated; } async function syncAutoListingDesktopTasks(plan = {}, actor = {}, options = {}) { const inputs = buildAutoListingDesktopTaskInputs(plan, actor); const tasks = []; for (const input of inputs) { tasks.push(await upsertDesktopExecutionTaskFromSource(input, actor, options)); } return tasks; } function buildListWhere(filters = {}, actor = {}) { const clauses = []; const params = []; if (actor.role !== 'admin') { clauses.push('(assigned_user_id = ? OR created_by_user_id = ?)'); params.push(normalizePositiveInteger(actor.userId || actor.id), normalizePositiveInteger(actor.userId || actor.id)); } else if (normalizePositiveInteger(filters.assignedUserId)) { clauses.push('assigned_user_id = ?'); params.push(normalizePositiveInteger(filters.assignedUserId)); } const status = normalizeTaskStatus(filters.status); if (status) { clauses.push('status = ?'); params.push(status); } if (filters.taskType) { clauses.push('task_type = ?'); params.push(normalizeTaskType(filters.taskType)); } return { sql: clauses.length ? `WHERE ${clauses.join(' AND ')}` : '', params }; } async function listDesktopExecutionTasks(filters = {}, actor = {}, options = {}) { const db = options.db || database.local; await ensureDesktopExecutionTaskSchema(db); const page = Math.max(1, Number.parseInt(filters.page, 10) || 1); const pageSize = Math.max(1, Math.min(100, Number.parseInt(filters.pageSize, 10) || 30)); const where = buildListWhere(filters, actor); const totalRows = await db.query(`SELECT COUNT(*) AS total FROM ${TASK_TABLE} ${where.sql}`, where.params); const rows = await db.query( `SELECT * FROM ${TASK_TABLE} ${where.sql} ORDER BY FIELD(status, 'running', 'claimed', 'queued', 'failed', 'succeeded', 'cancelled'), priority DESC, created_at DESC LIMIT ? OFFSET ?`, [...where.params, pageSize, (page - 1) * pageSize] ); return { rows: rows.map(mapTask), total: Number(totalRows[0]?.total || 0), page, pageSize }; } async function getDesktopExecutionSummary(filters = {}, actor = {}, options = {}) { const db = options.db || database.local; await ensureDesktopExecutionTaskSchema(db); const where = buildListWhere(filters, actor); const rows = await db.query( `SELECT status, COUNT(*) AS total FROM ${TASK_TABLE} ${where.sql} GROUP BY status`, where.params ); const counts = { queued: 0, claimed: 0, running: 0, succeeded: 0, failed: 0, cancelled: 0 }; rows.forEach((row) => { if (Object.hasOwn(counts, row.status)) counts[row.status] = Number(row.total || 0); }); return { counts, active: counts.queued + counts.claimed + counts.running, completed: counts.succeeded + counts.failed + counts.cancelled, }; } async function releaseExpiredClaims(db) { await db.update( `UPDATE ${TASK_TABLE} SET status = 'queued', claimed_client_id = NULL, claim_token_hash = NULL, claim_expires_at = NULL, progress_message = '上次领取已超时，等待重新领取', updated_at = NOW() WHERE status IN ('claimed', 'running') AND claim_expires_at IS NOT NULL AND claim_expires_at < NOW()`, [] ); } async function claimDesktopExecutionTask(input = {}, actor = {}, options = {}) { const db = options.db || database.local; await ensureDesktopExecutionTaskSchema(db); const assignedUserId = normalizePositiveInteger(actor.userId || actor.id); const clientId = normalizeText(input.clientId, 64).toLowerCase(); if (!assignedUserId || !clientId) { throw new DesktopExecutionTaskError('桌面任务领取身份不完整', 400, 'DESKTOP_TASK_CLAIM_IDENTITY_INVALID'); } const requestedTypes = (Array.isArray(input.taskTypes) ? input.taskTypes : []) .map(normalizeTaskType) .filter((value, index, values) => values.indexOf(value) === index); if (!requestedTypes.length) return { claimed: false, task: null }; await releaseExpiredClaims(db); const placeholders = requestedTypes.map(() => '?').join(', '); const candidates = await db.query( `SELECT id FROM ${TASK_TABLE} WHERE assigned_user_id = ? AND status = 'queued' AND task_type IN (${placeholders}) AND ( NULLIF(JSON_UNQUOTE(JSON_EXTRACT(input_json, '$.executorClientId')), '') IS NULL OR LOWER(JSON_UNQUOTE(JSON_EXTRACT(input_json, '$.executorClientId'))) = ? ) AND (scheduled_at IS NULL OR scheduled_at <= NOW()) AND (expires_at IS NULL OR expires_at > NOW()) ORDER BY priority DESC, created_at ASC LIMIT 5`, [assignedUserId, ...requestedTypes, clientId] ); const leaseSeconds = Math.max(60, Math.min(MAX_LEASE_SECONDS, normalizePositiveInteger(input.leaseSeconds, DEFAULT_LEASE_SECONDS))); for (const candidate of candidates) { const claimToken = crypto.randomBytes(32).toString('base64url'); const affected = await db.update( `UPDATE ${TASK_TABLE} SET status = 'claimed', claimed_client_id = ?, claim_token_hash = ?, claim_expires_at = DATE_ADD(NOW(), INTERVAL ? SECOND), attempt_count = attempt_count + 1, progress_stage = 'claimed', progress_message = '桌面客户端已领取', updated_at = NOW() WHERE id = ? AND status = 'queued'`, [clientId, hashClaimToken(claimToken), leaseSeconds, Number(candidate.id)] ); if (!affected) continue; const task = await getTaskById(candidate.id, { db }); await recordEvent(db, task, { eventType: 'claimed', fromStatus: 'queued', toStatus: 'claimed', actorType: 'desktop', actorId: assignedUserId, actorName: actor.name, clientId, message: '桌面客户端领取任务', }); return { claimed: true, claimToken, leaseSeconds, task }; } return { claimed: false, task: null }; } async function assertClaimedTask(taskId, input = {}, actor = {}, options = {}) { const db = options.db || database.local; const task = await getTaskById(taskId, { db }); if (!task) throw new DesktopExecutionTaskError('桌面任务不存在', 404, 'DESKTOP_TASK_NOT_FOUND'); const userId = normalizePositiveInteger(actor.userId || actor.id); const clientId = normalizeText(input.clientId, 64).toLowerCase(); if (task.assignedUserId !== userId || task.claimedClientId !== clientId) { throw new DesktopExecutionTaskError('任务不属于当前桌面会话', 403, 'DESKTOP_TASK_CLAIM_OWNER_MISMATCH'); } const rows = await db.query(`SELECT claim_token_hash FROM ${TASK_TABLE} WHERE id = ? LIMIT 1`, [task.id]); if (!input.claimToken || rows[0]?.claim_token_hash !== hashClaimToken(input.claimToken)) { throw new DesktopExecutionTaskError('任务领取凭证无效', 409, 'DESKTOP_TASK_CLAIM_TOKEN_INVALID'); } if (!ACTIVE_STATUSES.has(task.status)) { throw new DesktopExecutionTaskError('任务已结束，不能继续回传', 409, 'DESKTOP_TASK_ALREADY_FINISHED'); } return { db, task, clientId, userId }; } function normalizeProgress(input = {}) { const total = Math.max(0, Number.parseInt(input.total, 10) || 0); const current = Math.max(0, Number.parseInt(input.current, 10) || 0); const derived = total > 0 ? (Math.min(current, total) / total) * 100 : Number(input.percent || 0); return { stage: normalizeText(input.stage || 'running', 80), current, total, percent: Math.max(0, Math.min(100, Number.isFinite(derived) ? derived : 0)), message: normalizeText(input.message, 500), }; } async function updateDesktopExecutionProgress(taskId, input = {}, actor = {}, options = {}) { const context = await assertClaimedTask(taskId, input, actor, options); const progress = normalizeProgress(input); await context.db.update( `UPDATE ${TASK_TABLE} SET status = 'running', progress_stage = ?, progress_current = ?, progress_total = ?, progress_percent = ?, progress_message = ?, started_at = COALESCE(started_at, NOW()), claim_expires_at = GREATEST( COALESCE(claim_expires_at, NOW()), DATE_ADD(NOW(), INTERVAL ? SECOND) ), updated_at = NOW() WHERE id = ?`, [progress.stage, progress.current, progress.total, progress.percent, progress.message || null, DEFAULT_LEASE_SECONDS, context.task.id] ); return getTaskById(context.task.id, { db: context.db }); } async function finishDesktopExecutionTask(taskId, input = {}, actor = {}, outcome, options = {}) { const context = await assertClaimedTask(taskId, input, actor, options); const succeeded = outcome === 'succeeded'; const toStatus = succeeded ? 'succeeded' : 'failed'; const message = normalizeText(input.message || input.errorMessage, 1000); await context.db.update( `UPDATE ${TASK_TABLE} SET status = ?, progress_stage = ?, progress_percent = ?, progress_message = ?, result_json = ?, error_message = ?, completed_at = NOW(), claim_expires_at = NULL, updated_at = NOW() WHERE id = ?`, [ toStatus, succeeded ? 'completed' : 'failed', succeeded ? 100 : context.task.progress.percent, message || (succeeded ? '执行完成' : '执行失败'), JSON.stringify(normalizeJsonObject(input.result, '执行结果')), succeeded ? null : (message || '桌面任务执行失败'), context.task.id, ] ); const task = await getTaskById(context.task.id, { db: context.db }); await recordEvent(context.db, task, { eventType: toStatus, fromStatus: context.task.status, toStatus, actorType: 'desktop', actorId: context.userId, actorName: actor.name, clientId: context.clientId, message: message || (succeeded ? '桌面任务执行完成' : '桌面任务执行失败'), }); return task; } async function completeDesktopExecutionTask(taskId, input, actor, options) { return finishDesktopExecutionTask(taskId, input, actor, 'succeeded', options); } async function failDesktopExecutionTask(taskId, input, actor, options) { return finishDesktopExecutionTask(taskId, input, actor, 'failed', options); } async function cancelDesktopExecutionTask(taskId, actor = {}, options = {}) { const db = options.db || database.local; const task = await getTaskById(taskId, { db }); if (!task) throw new DesktopExecutionTaskError('桌面任务不存在', 404, 'DESKTOP_TASK_NOT_FOUND'); if (actor.role !== 'admin' && task.createdByUserId !== normalizePositiveInteger(actor.userId || actor.id)) { throw new DesktopExecutionTaskError('无权取消该任务', 403, 'DESKTOP_TASK_ACCESS_DENIED'); } if (TERMINAL_STATUSES.has(task.status)) return task; if (task.status !== 'queued') { throw new DesktopExecutionTaskError( '任务已被客户端领取，不能在总控端直接取消', 409, 'DESKTOP_TASK_CANCEL_NOT_ALLOWED' ); } await db.update( `UPDATE ${TASK_TABLE} SET status = 'cancelled', cancelled_at = NOW(), completed_at = NOW(), progress_stage = 'cancelled', progress_message = '总控已取消任务', claim_expires_at = NULL, updated_at = NOW() WHERE id = ? AND status = 'queued'`, [task.id] ); const updated = await getTaskById(task.id, { db }); await recordEvent(db, updated, { eventType: 'cancelled', fromStatus: task.status, toStatus: 'cancelled', actorType: 'control_center', actorId: actor.userId || actor.id, actorName: actor.name || actor.username, message: '总控取消任务', }); return updated; } async function retryDesktopExecutionTask(taskId, actor = {}, options = {}) { const db = options.db || database.local; const task = await getTaskById(taskId, { db }); if (!task) throw new DesktopExecutionTaskError('桌面任务不存在', 404, 'DESKTOP_TASK_NOT_FOUND'); if (actor.role !== 'admin' && task.createdByUserId !== normalizePositiveInteger(actor.userId || actor.id)) { throw new DesktopExecutionTaskError('无权重试该任务', 403, 'DESKTOP_TASK_ACCESS_DENIED'); } if (!['failed', 'cancelled'].includes(task.status)) { throw new DesktopExecutionTaskError('只有失败或已取消任务可以重试', 409, 'DESKTOP_TASK_RETRY_NOT_ALLOWED'); } await db.update( `UPDATE ${TASK_TABLE} SET status = 'queued', progress_stage = '', progress_current = 0, progress_total = 0, progress_percent = 0, progress_message = '等待桌面客户端领取', claimed_client_id = NULL, claim_token_hash = NULL, claim_expires_at = NULL, result_json = NULL, error_message = NULL, started_at = NULL, completed_at = NULL, cancelled_at = NULL, updated_at = NOW() WHERE id = ?`, [task.id] ); const updated = await getTaskById(task.id, { db }); await recordEvent(db, updated, { eventType: 'retried', fromStatus: task.status, toStatus: 'queued', actorType: 'control_center', actorId: actor.userId || actor.id, actorName: actor.name || actor.username, message: '总控重新下发任务', }); return updated; } function resetDesktopExecutionTaskStateForTests() { schemaReady = false; schemaPromise = null; } module.exports = { ACTIVE_STATUSES, DesktopExecutionTaskError, TASK_TYPES, buildTaskCode, buildAutoListingDesktopTaskInputs, cancelDesktopExecutionTask, claimDesktopExecutionTask, completeDesktopExecutionTask, createDesktopExecutionTask, ensureDesktopExecutionTaskSchema, failDesktopExecutionTask, getDesktopExecutionSummary, getTaskById, listDesktopExecutionTasks, mapTask, normalizeProgress, normalizeTaskType, resetDesktopExecutionTaskStateForTests, retryDesktopExecutionTask, syncAutoListingDesktopTasks, upsertDesktopExecutionTaskFromSource, updateDesktopExecutionProgress, };