88 lines
2.6 KiB
JavaScript
88 lines
2.6 KiB
JavaScript
import cron from 'node-cron';
|
||
import { cron_task_list } from '../config/cron_tasks.js';
|
||
import { get_flow_runner } from './flows/flow_registry.js';
|
||
|
||
const cron_jobs = [];
|
||
const running_task_name_set = new Set();
|
||
|
||
/**
|
||
* 启动参数开关（用于本地调试/冷启动后立即跑一次 cron）
|
||
* - 通过 VSCode/Cursor 的 launch.json 传入：--run_cron_now
|
||
* - 目的：避免等待 cron 表达式下一次触发（尤其是小时级任务）
|
||
*/
|
||
function has_argv_flag(flag_name) {
|
||
const name = String(flag_name || '').trim();
|
||
if (!name) return false;
|
||
return process.argv.includes(name);
|
||
}
|
||
|
||
function should_run_cron_now() {
|
||
return has_argv_flag('--run_cron_now');
|
||
}
|
||
|
||
async function run_cron_task(task) {
|
||
if (!task || !task.type) {
|
||
throw new Error('cron_task 缺少 type');
|
||
}
|
||
|
||
// 当前项目 cron 只允许跑 flow：任务入口集中，便于统一治理
|
||
if (task.type === 'flow') {
|
||
const run_flow = get_flow_runner(task.flow_name);
|
||
await run_flow(task.flow_payload || {});
|
||
return;
|
||
}
|
||
|
||
throw new Error(`cron_task type 不支持: ${task.type}`);
|
||
}
|
||
|
||
/**
|
||
* 统一的“防重复运行 + 执行 + 错误兜底”入口
|
||
* - 防止同一任务执行时间过长时，被下一次 cron 触发叠加执行
|
||
* - run_now 与定时触发复用同一套 guard，保证行为一致
|
||
*/
|
||
async function run_cron_task_with_guard(task_name, task) {
|
||
if (running_task_name_set.has(task_name)) {
|
||
// eslint-disable-next-line no-console
|
||
console.log('[cron] skip (already running)', { name: task_name });
|
||
return;
|
||
}
|
||
|
||
running_task_name_set.add(task_name);
|
||
try {
|
||
await run_cron_task(task);
|
||
} catch (error) {
|
||
console.warn('[cron] error', { task_name, error });
|
||
} finally {
|
||
running_task_name_set.delete(task_name);
|
||
}
|
||
}
|
||
|
||
export async function start_all_cron_tasks() {
|
||
const run_now = should_run_cron_now();
|
||
for (const task of cron_task_list) {
|
||
const task_name = task && task.name ? String(task.name) : 'cron_task';
|
||
|
||
// 先注册 cron（无论是否 run_now，都需要后续按表达式持续执行）
|
||
const job = cron.schedule(task.cron_expression, async () => {
|
||
await run_cron_task_with_guard(task_name, task);
|
||
});
|
||
console.log('job', { task_name, });
|
||
cron_jobs.push(job);
|
||
|
||
if (run_now) {
|
||
// 启动时额外立刻跑一次（仍走 guard，避免与 cron 触发撞车）
|
||
// eslint-disable-next-line no-console
|
||
console.log('[cron] run_now', { task_name });
|
||
await run_cron_task_with_guard(task_name, task);
|
||
}
|
||
}
|
||
}
|
||
|
||
export function stop_all_cron_tasks() {
|
||
for (const job of cron_jobs) {
|
||
job.stop();
|
||
}
|
||
cron_jobs.length = 0;
|
||
running_task_name_set.clear();
|
||
}
|