import { logError, logInfo, logWarn } from '../../utils/logger.js' import { createHttpError } from '../../utils/http.js' import { getCloudtentaclesHealthJob, getScheduledJobsConfig, } from './config-service.js' import { runCloudtentaclesHealthJob } from './cloudtentacles-health-job.js' type JsonObject = Record const timers = new Map() const jobStates = new Map() export function startScheduledJobs() { reloadScheduledJobs() } export function stopScheduledJobs() { for (const timer of timers.values()) { clearTimeout(timer) } timers.clear() } export function reloadScheduledJobs() { stopScheduledJobs() const config = getScheduledJobsConfig() if (config.enabled === false) { logInfo('[scheduler]', '定时任务系统已停用', {}) return } for (const job of config.jobs) { if (job.enabled !== true) { updateJobState(job.id, { enabled: false, running: false, nextRunAt: '', }) continue } scheduleJob(job, 1000) } } export function getScheduledJobRuntimeStates() { return Array.from(jobStates.values()) } export async function runScheduledJobNow(jobId: unknown) { const config = getScheduledJobsConfig() const job = config.jobs.find((item) => String(item.id || '').trim() === String(jobId || '').trim()) if (!job) { throw createHttpError('定时任务不存在', { statusCode: 404, errorCode: 'scheduled_job_not_found', }) } const currentTimer = timers.get(job.id) if (currentTimer) { clearTimeout(currentTimer) timers.delete(job.id) } const result = await runJob(job, { manual: true }) if (getScheduledJobsConfig().enabled !== false) { const latestJob = getCloudtentaclesHealthJob(getScheduledJobsConfig()) if (latestJob.enabled === true) { scheduleJob(latestJob, Math.max(60, Number(latestJob.intervalSeconds || 300)) * 1000) } } return result } function scheduleJob(job: JsonObject, delayMs: number) { const jobId = String(job.id || '').trim() if (!jobId) { return } const nextRunAt = new Date(Date.now() + delayMs).toISOString() updateJobState(jobId, { enabled: true, nextRunAt, }) const timer = setTimeout(() => { timers.delete(jobId) void runJob(job).finally(() => { const latestJob = getCloudtentaclesHealthJob(getScheduledJobsConfig()) if (latestJob.enabled === true) { scheduleJob(latestJob, Math.max(60, Number(latestJob.intervalSeconds || 300)) * 1000) } }) }, delayMs) timers.set(jobId, timer) } async function runJob(job: JsonObject, { manual = false }: { manual?: boolean } = {}) { const jobId = String(job.id || '').trim() const startedAt = new Date().toISOString() if (!jobId) { return null } if (jobStates.get(jobId)?.running) { logWarn('[scheduler]', '定时任务仍在运行,跳过本次执行', { jobId }) return { skipped: true, reason: 'running', } } updateJobState(jobId, { enabled: job.enabled === true, running: true, lastRunAt: startedAt, lastStatus: 'running', lastMessage: '执行中', }) try { const result = await dispatchJob(job) updateJobState(jobId, { running: false, lastFinishedAt: new Date().toISOString(), lastStatus: result?.status || 'ok', lastMessage: result?.message || '执行完成', lastAsset: typeof result?.asset === 'number' ? result.asset : null, lastThreshold: typeof result?.threshold === 'number' ? result.threshold : null, lastAccounts: Array.isArray(result?.accounts) ? result.accounts : [], lastAccountCount: Number(result?.accountCount || 0), lastCheckedCount: Number(result?.checkedCount || 0), lastOkCount: Number(result?.okCount || 0), lastLowAssetCount: Number(result?.lowAssetCount || 0), lastFailedCount: Number(result?.failedCount || 0), lastManual: manual, }) logInfo('[scheduler]', '定时任务执行完成', { jobId, type: job.type, status: result?.status || 'ok', }) return result } catch (error) { const message = error instanceof Error ? error.message : String(error || '定时任务执行失败') updateJobState(jobId, { running: false, lastFinishedAt: new Date().toISOString(), lastStatus: 'failed', lastMessage: message, lastManual: manual, }) logError('[scheduler]', '定时任务执行失败', { jobId, type: job.type, error, }) return { ok: false, status: 'failed', message, } } } function dispatchJob(job: JsonObject) { if (job.type === 'cloudtentacles_health') { return runCloudtentaclesHealthJob(job) } throw createHttpError(`不支持的定时任务类型:${job.type}`, { statusCode: 400, errorCode: 'unsupported_scheduled_job_type', }) } function updateJobState(jobId: string, patch: JsonObject) { const current = jobStates.get(jobId) || { id: jobId, enabled: false, running: false, lastRunAt: '', lastFinishedAt: '', nextRunAt: '', lastStatus: 'pending', lastMessage: '', lastAsset: null, lastThreshold: null, lastAccounts: [], lastAccountCount: 0, lastCheckedCount: 0, lastOkCount: 0, lastLowAssetCount: 0, lastFailedCount: 0, lastManual: false, } jobStates.set(jobId, { ...current, ...patch, }) }