简单的定时任务--ok

2026-05-14 17:59:45 +08:00
parent b6c493d5e7
commit f7f3289d04
19 changed files with 954 additions and 5 deletions
@@ -9,6 +9,7 @@ import open91Router from './routes/open-91.js'
 import webhooksRouter from './routes/webhooks.js'
 import { ensureAdminUsersBootstrapped } from './services/admin/admin-auth-service.js'
 import { ensureFulfillmentCatalogBootstrapped } from './services/bootstrap/fulfillment-bootstrap-service.js'
+import { startScheduledJobs, stopScheduledJobs } from './services/scheduler/scheduler-service.js'
 import { closeLocalOcrWorker, warmupLocalOcrWorker } from './services/session/ocr.js'
 import { closeAllTencentBrowserSessions, warmupTencentBrowser } from './services/session/session.js'
 import { buildSuccessPayload } from './utils/http.js'
@@ -162,6 +163,7 @@ async function bootstrapCoreServices() {
        attempt: startupState.core.attemptCount,
      })

+      startScheduledJobs()
      void bootstrapBrowser()
      void bootstrapOcr()
      break
@@ -358,6 +360,8 @@ async function shutdown(signal) {
  startupState.phase = 'shutting_down'
  logInfo('[shutdown]', `received ${signal}, closing browser sessions and HTTP server`)

+  stopScheduledJobs()
+
  try {
    await closeAllTencentBrowserSessions({ markClosed: true })
  } catch (error) {
@@ -12,6 +12,7 @@ import {
  getAdminCloudtentaclesBindUrl,
  getAdminCloudtentaclesCategories,
  getAdminNotificationConfig,
+  getAdminScheduledJobsConfig,
  getAdminKuaishouCloudFulfillmentConfig,
  getAdminKuaishouEticketSourceConfig,
  getAdminCloudtentaclesKnapsack,
@@ -29,6 +30,7 @@ import {
  queryAdminKuaishouEticketDetail,
  retryAdminNinetyoneOrder,
  runAdminCloudtentaclesFullFlow,
+  runAdminScheduledJobNow,
  sendAdminCloudtentaclesSmsCode,
  testAdminNotification,
  testAdminCloudtentaclesLogin,
@@ -36,6 +38,7 @@ import {
  updateAdminKuaishouEticketSourceConfig,
  updateAdminCloudtentaclesSourceConfig,
  updateAdminNotificationConfig,
+  updateAdminScheduledJobsConfig,
  updateAdminAgisoShopConfigs,
  updateAdminFulfillmentBindingConfigs,
  verifyAdminCloudtentaclesLoginCode,
@@ -50,6 +53,7 @@ import { createJsonHandler, requireAdminRoles } from './shared.js'
 /** @typedef {import('../../types/admin-route-inputs.js').AdminKuaishouEticketSourceConfigRouteBody} AdminKuaishouEticketSourceConfigRouteBody */
 /** @typedef {import('../../types/admin-route-inputs.js').AdminNotificationConfigRouteBody} AdminNotificationConfigRouteBody */
 /** @typedef {import('../../types/admin-route-inputs.js').AdminNotificationTestRouteBody} AdminNotificationTestRouteBody */
+/** @typedef {import('../../types/admin-route-inputs.js').AdminScheduledJobsConfigRouteBody} AdminScheduledJobsConfigRouteBody */
 /** @typedef {import('../../types/admin-route-inputs.js').AdminKuaishouEticketDetailQueryRouteBody} AdminKuaishouEticketDetailQueryRouteBody */
 /** @typedef {import('../../types/admin-route-inputs.js').AdminKuaishouEticketConsumeRouteBody} AdminKuaishouEticketConsumeRouteBody */
 /** @typedef {import('../../types/admin-route-inputs.js').AdminKuaishouEticketShopInfoRouteBody} AdminKuaishouEticketShopInfoRouteBody */
@@ -154,6 +158,58 @@ router.post('/platform-config/notifications/test', createJsonHandler(
  },
 ))

+router.get('/platform-config/scheduled-jobs', createJsonHandler(
+  () => getAdminScheduledJobsConfig(),
+  {
+    successMessage: 'ok',
+    errorMessage: '读取定时任务配置失败',
+    scope: '[admin/platform-config/scheduled-jobs]',
+  },
+))
+
+router.post('/platform-config/scheduled-jobs', createJsonHandler(
+  (req) => updateAdminScheduledJobsConfig(/** @type {AdminScheduledJobsConfigRouteBody} */ (req.body)),
+  {
+    successMessage: '定时任务配置已保存',
+    errorMessage: '保存定时任务配置失败',
+    scope: '[admin/platform-config/scheduled-jobs]',
+    audit: (_req, data) => {
+      const result = /** @type {{ filePath?: string, source?: { enabled?: boolean, jobs?: unknown[] } }} */ (data)
+      return {
+        action: 'platform_scheduled_jobs_updated',
+        targetType: 'platform_config',
+        targetId: 'scheduled_jobs',
+        data: {
+          filePath: String(result.filePath || '').trim(),
+          enabled: Boolean(result.source?.enabled),
+          jobCount: Array.isArray(result.source?.jobs) ? result.source.jobs.length : 0,
+        },
+      }
+    },
+  },
+))
+
+router.post('/platform-config/scheduled-jobs/:id/run', createJsonHandler(
+  (req) => runAdminScheduledJobNow((/** @type {AdminEntityRouteParams} */ (req.params)).id),
+  {
+    successMessage: '定时任务已执行',
+    errorMessage: '执行定时任务失败',
+    scope: '[admin/platform-config/scheduled-jobs/:id/run]',
+    audit: (req, data) => {
+      const result = /** @type {{ result?: { status?: string, message?: string } }} */ (data)
+      return {
+        action: 'platform_scheduled_job_run',
+        targetType: 'platform_config',
+        targetId: String((/** @type {AdminEntityRouteParams} */ (req.params)).id || '').trim(),
+        data: {
+          status: String(result.result?.status || '').trim(),
+          message: String(result.result?.message || '').trim(),
+        },
+      }
+    },
+  },
+))
+
 router.get('/platform-config/kuaishou-eticket-source', createJsonHandler(
  () => getAdminKuaishouEticketSourceConfig(),
  {
@@ -6,10 +6,21 @@ import {
  saveNotificationConfig,
 } from '../../notification/config-service.js'
 import { sendInternalNotification } from '../../notification/notification-service.js'
+import {
+  getScheduledJobsConfig,
+  getScheduledJobsFilePath,
+  saveScheduledJobsConfig,
+} from '../../scheduler/config-service.js'
+import {
+  getScheduledJobRuntimeStates,
+  reloadScheduledJobs,
+  runScheduledJobNow,
+} from '../../scheduler/scheduler-service.js'
 import { maskSecret } from './mappers.js'

 /** @typedef {import('../../../types/admin-write-inputs.js').AdminNotificationConfigInput} AdminNotificationConfigInput */
 /** @typedef {import('../../../types/admin-write-inputs.js').AdminNotificationTestInput} AdminNotificationTestInput */
+/** @typedef {import('../../../types/admin-write-inputs.js').AdminScheduledJobsConfigInput} AdminScheduledJobsConfigInput */

 export function getAdminNotificationConfig() {
  const config = getNotificationConfig()
@@ -40,6 +51,37 @@ export async function testAdminNotification(payload = /** @type {AdminNotificati
  })
 }

+export function getAdminScheduledJobsConfig() {
+  const config = getScheduledJobsConfig()
+
+  return {
+    filePath: getScheduledJobsFilePath(),
+    source: mapAdminScheduledJobsConfig(config),
+    runtime: getScheduledJobRuntimeStates(),
+  }
+}
+
+/** @param {AdminScheduledJobsConfigInput} [payload] */
+export function updateAdminScheduledJobsConfig(payload = /** @type {AdminScheduledJobsConfigInput} */ ({})) {
+  const saved = saveScheduledJobsConfig(payload)
+  reloadScheduledJobs()
+
+  return {
+    filePath: getScheduledJobsFilePath(),
+    source: mapAdminScheduledJobsConfig(saved),
+    runtime: getScheduledJobRuntimeStates(),
+  }
+}
+
+export async function runAdminScheduledJobNow(jobId) {
+  const result = await runScheduledJobNow(jobId)
+
+  return {
+    result,
+    runtime: getScheduledJobRuntimeStates(),
+  }
+}
+
 function mapAdminNotificationConfig(config = {}) {
  const bark = config.channels?.bark || {}

@@ -60,3 +102,19 @@ function mapAdminNotificationConfig(config = {}) {
    },
  }
 }
+
+function mapAdminScheduledJobsConfig(config = {}) {
+  return {
+    enabled: config.enabled !== false,
+    jobs: (Array.isArray(config.jobs) ? config.jobs : []).map((item) => ({
+      id: String(item.id || '').trim(),
+      type: String(item.type || '').trim(),
+      enabled: item.enabled === true,
+      intervalSeconds: Number(item.intervalSeconds || 300),
+      cooldownSeconds: Number(item.cooldownSeconds || 1800),
+      config: {
+        assetThreshold: Number(item.config?.assetThreshold || 500),
+      },
+    })),
+  }
+}
@@ -54,4 +54,7 @@ export {
  getAdminNotificationConfig,
  updateAdminNotificationConfig,
  testAdminNotification,
+  getAdminScheduledJobsConfig,
+  updateAdminScheduledJobsConfig,
+  runAdminScheduledJobNow,
 } from './notification-service.js'
@@ -82,6 +82,7 @@ export function notifyCloudtentaclesAuthExpired({
  pathname = '',
  errorCode = '',
  message = '',
+  cooldownSeconds = 600,
 } = {}) {
  return notifyInternalSafely({
    title: 'cloudtentacles 登录已过期',
@@ -92,6 +93,25 @@ export function notifyCloudtentaclesAuthExpired({
    ].join('\n'),
    category: 'cloudtentacles_auth_expired',
    cooldownKey: ['cloudtentacles_auth_expired', pathname, errorCode].join(':'),
+    cooldownMs: Number(cooldownSeconds || 600) * 1000,
+  })
+}
+
+export function notifyCloudtentaclesAssetLow({
+  asset = 0,
+  threshold = 500,
+  cooldownSeconds = 1800,
+} = {}) {
+  return notifyInternalSafely({
+    title: '快手 Cloud 余额低于阈值',
+    body: [
+      `当前余额：${Number(asset || 0)}`,
+      `提醒阈值：${Number(threshold || 0)}`,
+      '请及时补充 cloudtentacles 余额，避免自动履约失败。',
+    ].join('\n'),
+    category: 'cloudtentacles_asset_low',
+    cooldownKey: ['cloudtentacles_asset_low', threshold].join(':'),
+    cooldownMs: Number(cooldownSeconds || 1800) * 1000,
  })
 }

@@ -0,0 +1,58 @@
+// @ts-check
+
+import { getCloudtentaclesAsset } from '../platforms/cloudtentacles/catalog-service.js'
+import { getCloudtentaclesSourceConfig } from '../platforms/cloudtentacles/source-config-service.js'
+import { getCloudtentaclesSessionState } from '../platforms/cloudtentacles/session-state-service.js'
+import {
+  notifyCloudtentaclesAssetLow,
+  notifyCloudtentaclesAuthExpired,
+} from '../notification/domain-notifications.js'
+
+export async function runCloudtentaclesHealthJob(job) {
+  const source = getCloudtentaclesSourceConfig()
+  const session = getCloudtentaclesSessionState()
+  const token = String(session.token || '').trim()
+  const threshold = Number(job.config?.assetThreshold || 500)
+  const cooldownSeconds = Number(job.cooldownSeconds || 1800)
+
+  if (!token) {
+    await notifyCloudtentaclesAuthExpired({
+      pathname: '/user/get_asset',
+      errorCode: 'cloudtentacles_token_missing',
+      message: '当前 cloudtentacles 没有可用 token，请到后台重新登录',
+      cooldownSeconds,
+    })
+
+    return {
+      ok: false,
+      status: 'auth_missing',
+      message: 'cloudtentacles token 缺失',
+      asset: null,
+      threshold,
+    }
+  }
+
+  const assetResult = await getCloudtentaclesAsset({
+    baseUrl: String(session.baseUrl || source.baseUrl || '').trim() || 'https://123.207.217.176',
+    token,
+    deviceId: String(session.deviceId || source.deviceId || '-').trim() || '-',
+    deviceType: Number(session.deviceType ?? source.deviceType ?? 0),
+  })
+  const asset = Number(assetResult.asset || 0)
+
+  if (asset < threshold) {
+    await notifyCloudtentaclesAssetLow({
+      asset,
+      threshold,
+      cooldownSeconds,
+    })
+  }
+
+  return {
+    ok: true,
+    status: asset < threshold ? 'asset_low' : 'ok',
+    message: asset < threshold ? `余额 ${asset} 低于阈值 ${threshold}` : `余额 ${asset} 正常`,
+    asset,
+    threshold,
+  }
+}
@@ -0,0 +1,122 @@
+// @ts-check
+
+import fs from 'node:fs'
+import path from 'node:path'
+
+import { PROJECT_ROOT } from '../../config/runtime.js'
+
+const SCHEDULED_JOBS_FILE_PATH = path.join(PROJECT_ROOT, 'data', 'scheduled-jobs.json')
+const CLOUDTENTACLES_HEALTH_JOB_ID = 'cloudtentacles-health'
+
+export function getScheduledJobsFilePath() {
+  return SCHEDULED_JOBS_FILE_PATH
+}
+
+export function getScheduledJobsConfig() {
+  return loadScheduledJobsConfigFromFile()
+}
+
+export function saveScheduledJobsConfig(rawValue) {
+  const normalized = normalizeScheduledJobsConfig(rawValue)
+  fs.mkdirSync(path.dirname(SCHEDULED_JOBS_FILE_PATH), { recursive: true })
+  fs.writeFileSync(SCHEDULED_JOBS_FILE_PATH, `${JSON.stringify(normalized, null, 2)}\n`, 'utf8')
+  return normalized
+}
+
+export function getCloudtentaclesHealthJob(config = getScheduledJobsConfig()) {
+  return (Array.isArray(config.jobs) ? config.jobs : [])
+    .find((item) => String(item.id || '').trim() === CLOUDTENTACLES_HEALTH_JOB_ID)
+    || createDefaultCloudtentaclesHealthJob()
+}
+
+function loadScheduledJobsConfigFromFile() {
+  if (!fs.existsSync(SCHEDULED_JOBS_FILE_PATH)) {
+    return createDefaultScheduledJobsConfig()
+  }
+
+  try {
+    const rawText = fs.readFileSync(SCHEDULED_JOBS_FILE_PATH, 'utf8')
+    return normalizeScheduledJobsConfig(JSON.parse(rawText))
+  } catch {
+    return createDefaultScheduledJobsConfig()
+  }
+}
+
+function normalizeScheduledJobsConfig(rawValue) {
+  const source = isPlainObject(rawValue) ? rawValue : {}
+  const rawJobs = Array.isArray(source.jobs) ? source.jobs : []
+  const jobs = rawJobs.map((item) => normalizeScheduledJob(item)).filter(Boolean)
+  const hasCloudtentaclesHealth = jobs.some((item) => item.id === CLOUDTENTACLES_HEALTH_JOB_ID)
+
+  if (!hasCloudtentaclesHealth) {
+    jobs.push(createDefaultCloudtentaclesHealthJob())
+  }
+
+  return {
+    enabled: typeof source.enabled === 'boolean' ? source.enabled : true,
+    jobs,
+  }
+}
+
+function normalizeScheduledJob(rawValue) {
+  if (!isPlainObject(rawValue)) {
+    return null
+  }
+
+  const type = String(rawValue.type || '').trim()
+  if (type !== 'cloudtentacles_health') {
+    return null
+  }
+
+  return normalizeCloudtentaclesHealthJob(rawValue)
+}
+
+function normalizeCloudtentaclesHealthJob(rawValue) {
+  const config = isPlainObject(rawValue.config) ? rawValue.config : {}
+
+  return {
+    id: CLOUDTENTACLES_HEALTH_JOB_ID,
+    type: 'cloudtentacles_health',
+    enabled: rawValue.enabled === true,
+    intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 300, 60, 86400),
+    cooldownSeconds: normalizeRangeInteger(rawValue.cooldownSeconds, 1800, 60, 86400),
+    config: {
+      assetThreshold: normalizeRangeInteger(config.assetThreshold, 500, 0, 999999),
+    },
+  }
+}
+
+function createDefaultScheduledJobsConfig() {
+  return {
+    enabled: true,
+    jobs: [
+      createDefaultCloudtentaclesHealthJob(),
+    ],
+  }
+}
+
+function createDefaultCloudtentaclesHealthJob() {
+  return {
+    id: CLOUDTENTACLES_HEALTH_JOB_ID,
+    type: 'cloudtentacles_health',
+    enabled: false,
+    intervalSeconds: 300,
+    cooldownSeconds: 1800,
+    config: {
+      assetThreshold: 500,
+    },
+  }
+}
+
+function normalizeRangeInteger(value, fallback, min, max) {
+  const parsed = Number(value)
+  if (!Number.isInteger(parsed)) {
+    return fallback
+  }
+
+  return Math.min(max, Math.max(min, parsed))
+}
+
+function isPlainObject(value) {
+  return Object.prototype.toString.call(value) === '[object Object]'
+}
@@ -0,0 +1,189 @@
+// @ts-check
+
+import { logError, logInfo, logWarn } from '../../utils/logger.js'
+import {
+  getCloudtentaclesHealthJob,
+  getScheduledJobsConfig,
+} from './config-service.js'
+import { runCloudtentaclesHealthJob } from './cloudtentacles-health-job.js'
+
+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) {
+  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, delayMs) {
+  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, { manual = false } = {}) {
+  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) {
+  if (job.type === 'cloudtentacles_health') {
+    return runCloudtentaclesHealthJob(job)
+  }
+
+  throw new Error(`不支持的定时任务类型：${job.type}`)
+}
+
+function updateJobState(jobId, patch) {
+  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,
+  })
+}
@@ -42,6 +42,10 @@ export {}
 * @typedef {import('./admin-write-inputs.js').AdminNotificationTestInput} AdminNotificationTestRouteBody
 */

+/**
+ * @typedef {import('./admin-write-inputs.js').AdminScheduledJobsConfigInput} AdminScheduledJobsConfigRouteBody
+ */
+
 /**
 * @typedef {import('./admin-write-inputs.js').AdminAgisoShopConfigSaveInput} AdminAgisoShopConfigRouteBody
 */
@@ -150,6 +150,26 @@ export {}
 * }} AdminNotificationTestInput
 */

+/**
+ * @typedef {{
+ *   id?: string
+ *   type?: string
+ *   enabled?: boolean
+ *   intervalSeconds?: number | string
+ *   cooldownSeconds?: number | string
+ *   config?: {
+ *     assetThreshold?: number | string
+ *   }
+ * }} AdminScheduledJobInput
+ */
+
+/**
+ * @typedef {{
+ *   enabled?: boolean
+ *   jobs?: AdminScheduledJobInput[]
+ * }} AdminScheduledJobsConfigInput
+ */
+
 /**
 * @typedef {{
  *   baseUrl?: string