order_site/apps/backend/src/services/scheduler/scheduler-service.ts

import { logError, logInfo, logWarn } from '../../utils/logger.js'
import { createHttpError } from '../../utils/http.js'
import type { JsonObject } from '../../types/json.js'
import { getScheduledJobById, getScheduledJobsConfig } from './config-service.js'
import { runCloudtentaclesHealthJob } from './cloudtentacles-health-job.js'
import { runDepositUnfreezeJob } from './deposit-unfreeze-job.js'
import { runWorkOrderTimeoutJob } from './work-order-timeout-job.js'

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 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) {
    rescheduleJob(job)
  }
  return result
}

function rescheduleJob(job: JsonObject) {
  const jobId = String(job.id || '').trim()
  if (!jobId) return
  const latestJob = getScheduledJobById(jobId)
  if (!latestJob || latestJob.enabled !== true) return
  const intervalMs = Math.max(60, Number(latestJob.intervalSeconds || 60)) * 1000
  scheduleJob(latestJob, intervalMs)
}

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(() => {
      rescheduleJob(job)
    })
  }, 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)
    const summary = result && typeof result === 'object' ? (result as JsonObject) : {}
    updateJobState(jobId, {
      running: false,
      lastFinishedAt: new Date().toISOString(),
      lastStatus: String(summary.status || 'ok'),
      lastMessage: String(summary.message || '执行完成'),
      lastAsset: typeof summary.asset === 'number' ? summary.asset : null,
      lastThreshold: typeof summary.threshold === 'number' ? summary.threshold : null,
      lastAccounts: Array.isArray(summary.accounts) ? summary.accounts : [],
      lastAccountCount: Number(summary.accountCount || 0),
      lastCheckedCount: Number(summary.checkedCount || 0),
      lastOkCount: Number(summary.okCount || 0),
      lastLowAssetCount: Number(summary.lowAssetCount || 0),
      lastFailedCount: Number(summary.failedCount || 0),
      lastManual: manual,
    })
    logInfo('[scheduler]', '定时任务执行完成', {
      jobId,
      type: job.type,
      status: String(summary.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)
  }
  if (job.type === 'work_order_timeout') {
    return runWorkOrderTimeoutJob(job)
  }
  if (job.type === 'deposit_unfreeze') {
    return runDepositUnfreezeJob(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,
  })
}