后端迁移定时任务服务

2026-05-21 16:34:06 +08:00
parent 1c760f1dd3
commit f130bdba78
4 changed files with 32 additions and 22 deletions
@@ -0,0 +1,189 @@
+import { logError, logInfo, logWarn } from '../../utils/logger.js'
+import {
+  getCloudtentaclesHealthJob,
+  getScheduledJobsConfig,
+} from './config-service.js'
+import { runCloudtentaclesHealthJob } from './cloudtentacles-health-job.js'
+
+type JsonObject = Record<string, any>
+
+const timers = new Map<string, NodeJS.Timeout>()
+const jobStates = new Map<string, JsonObject>()
+
+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 new Error('定时任务不存在')
+  }
+
+  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,
+      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 new Error(`不支持的定时任务类型：${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,
+    lastManual: false,
+  }
+  jobStates.set(jobId, {
+    ...current,
+    ...patch,
+  })
+}