import type { ScheduleGroup, ScheduleKind, ScheduleSpec, ScheduleTrigger } from "@umt/contracts"; import type { ServerContext } from "../types.ts"; // 定时任务注册表(进程内):模块在 start() 里登记 {key, defaults, run},schedules 模块的调度器 // 把定义同步进 scheduled_job 表(不覆盖用户改过的开关/调度),按 next_run_at 调用 run。 // 各模块只 import 本文件(不依赖调度器实现);没装 schedules 模块时登记只是一张内存表,不会执行。 export interface ScheduledJobDefaults { name: string; description?: string; group: ScheduleGroup; schedule: ScheduleSpec; /** 缺省 true */ enabled?: boolean; /** 任务参数(看板可改;task 浅合并,agent_check 整体替换) */ config?: Record; /** false = 占位(如备份未配置),不执行、不可开启 */ available?: boolean; /** 首次登记时何时第一次运行:now(缺省,尽快)/ after_interval(一个周期后) */ firstRun?: "now" | "after_interval"; } export interface JobRunContext { jobId: string; /** scheduled_job_run.id */ runId: string; trigger: Exclude; now: Date; /** 当前任务参数(表里的 config) */ config: Record; /** * 时间槽:按调度周期取整的稳定串(interval:floor(now/everyMs);cron:本次计划时间)。 * 入队幂等键用它拼,多实例 / 重复触发同槽只入队一次;立即执行用 manual:。 */ slot: string; } export interface JobRunResult { /** 缺省 ok;业务侧主动放弃(如只读对账模式)返回 skipped */ status?: "ok" | "skipped"; summary?: string | null; agentRunId?: string | null; } export type JobRunFn = (c: JobRunContext) => Promise; export interface ScheduledJobDef { /** 稳定 key(表主键),如 feeds.poll */ key: string; kind?: ScheduleKind; defaults: ScheduledJobDefaults; run: JobRunFn; } export interface ScheduleRegistry { /** 每次登记 +1,调度器据此判断是否需要重新同步到表 */ readonly version: number; get(key: string): ScheduledJobDef | undefined; list(): ScheduledJobDef[]; register(def: ScheduledJobDef): void; unregister(key: string): void; } class MemoryRegistry implements ScheduleRegistry { version = 0; private defs = new Map(); get(key: string) { return this.defs.get(key); } list() { return [...this.defs.values()]; } register(def: ScheduledJobDef) { if (!/^[a-z][a-z0-9_.:-]{1,120}$/.test(def.key)) throw new Error(`定时任务 key 不合法:${def.key}`); this.defs.set(def.key, def); this.version += 1; } unregister(key: string) { if (this.defs.delete(key)) this.version += 1; } } const registries = new WeakMap(); /** 取 ctx 对应的注册表(同一 ServerContext 共用一张) */ export function scheduleRegistry(ctx: Pick): ScheduleRegistry { const k = ctx as object; let r = registries.get(k); if (!r) { r = new MemoryRegistry(); registries.set(k, r); } return r; } /** 模块 start() 里调用:登记一个周期任务(同 key 重复登记以最后一次为准) */ export function registerScheduledJob(ctx: Pick, def: ScheduledJobDef): void { scheduleRegistry(ctx).register(def); }