feat: 新增电工APP端与定时任务管理功能
- 新增电工APP独立鉴权中间件与登录注册接口 - 新增定时任务管理模块，支持任务配置、执行日志与手动触发 - 新增运维费定价与电工派单成本价目管理 - 新增电工资质证照管理与到期提醒定时任务 - 完善数据库实体与DTO校验，优化权限控制
This commit is contained in:
andyesfly committed
1 parent
1803ad679c
commit
f4454ad014
57 files changed
+9388
-200
No files matched your search
@@ -0,0 +1,429 @@
|
||||
import { Provide, App, Scope, ScopeEnum } from "@midwayjs/core";
|
||||
import { Application } from "@midwayjs/koa";
|
||||
import { InjectRepository } from "@midwayjs/sequelize";
|
||||
import { CronJob, CronTime } from "cron";
|
||||
import { Op } from "sequelize";
|
||||
import dayjs from "dayjs";
|
||||
import SysJob from "../entity/sys_job.entity";
|
||||
import SysJobLog from "../entity/sys_job_log.entity";
|
||||
import { LogService } from "./log.service";
|
||||
import { BuiltinJob, JobHandlerContext, JobRuntime, getBuiltinJob } from "../task/job_registry";
|
||||
|
||||
/** 任务执行结果（一次尝试） */
|
||||
export interface JobRunResult {
|
||||
status: number; // 1成功 2失败 3跳过
|
||||
message: string; // 结果摘要
|
||||
costMs: number; // 本次尝试耗时
|
||||
attempt: number; // 本次是第几次尝试
|
||||
}
|
||||
|
||||
/** 调度时区（固定上海时区） */
|
||||
const TIMEZONE = "Asia/Shanghai";
|
||||
/** 异常堆栈最大存储长度 */
|
||||
const MAX_ERROR_STACK = 4000;
|
||||
|
||||
/**
|
||||
* 自研轻量调度器：任务清单来自代码注册表，调度配置持久化在 sys_job，
|
||||
* 页面修改 cron / 启停后通过 reload 即时生效（无需重启）。
|
||||
*/
|
||||
// 调度器必须是进程内单例：定时器注册表要能被「启动注册」和「接口重载」共享
|
||||
@Provide()
|
||||
@Scope(ScopeEnum.Singleton)
|
||||
export class JobSchedulerService {
|
||||
@App()
|
||||
app!: Application;
|
||||
|
||||
@InjectRepository(SysJob)
|
||||
jobModel!: typeof SysJob;
|
||||
|
||||
@InjectRepository(SysJobLog)
|
||||
jobLogModel!: typeof SysJobLog;
|
||||
|
||||
/** jobKey → 定时器 */
|
||||
private timers = new Map<string, CronJob>();
|
||||
/** jobKey → 待执行的重试定时器 */
|
||||
private retryTimers = new Map<string, NodeJS.Timeout>();
|
||||
/** jobKey 重入锁 */
|
||||
private running = new Set<string>();
|
||||
|
||||
/** 私有日志适配：注册表与任务经它输出日志，沿用仓库内 console 风格 */
|
||||
private readonly jobLogger = {
|
||||
info: (m: string, ...a: any[]) => console.log(`[定时任务] ${m}`, ...a),
|
||||
warn: (m: string, ...a: any[]) => console.warn(`[定时任务] ${m}`, ...a),
|
||||
error: (m: string, ...a: any[]) => console.error(`[定时任务] ${m}`, ...a)
|
||||
};
|
||||
|
||||
/** 应用启动时加载全部启用任务并注册调度；失败自动重试，最多 3 次，不阻塞启动 */
|
||||
async start(): Promise<void> {
|
||||
const maxAttempts = 3;
|
||||
|
||||
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
|
||||
try {
|
||||
const jobs = await this.jobModel.findAll({ where: { status: 1 } });
|
||||
jobs.forEach(job => this.register(job));
|
||||
this.jobLogger.info(`调度器启动完成，已加载 ${jobs.length} 个任务`);
|
||||
return;
|
||||
} catch (err) {
|
||||
if (attempt < maxAttempts) {
|
||||
this.jobLogger.error(`调度器启动失败（第 ${attempt} 次尝试），5 秒后重试:`, err);
|
||||
await new Promise(resolve => setTimeout(resolve, 5000));
|
||||
} else {
|
||||
// 3 次仍失败仅记录，不抛出，避免阻塞应用启动
|
||||
console.error("[定时任务] 调度器启动失败，已达最大重试次数，应用继续启动:", err);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** 停止全部定时器（应用停止时调用） */
|
||||
async stop(): Promise<void> {
|
||||
for (const timer of this.timers.values()) {
|
||||
try {
|
||||
timer.stop();
|
||||
} catch (err) {
|
||||
console.warn("[定时任务] 停止定时器失败:", err);
|
||||
}
|
||||
}
|
||||
this.timers.clear();
|
||||
|
||||
for (const retryTimer of this.retryTimers.values()) {
|
||||
clearTimeout(retryTimer);
|
||||
}
|
||||
this.retryTimers.clear();
|
||||
}
|
||||
|
||||
/** 单个任务重新加载：停旧 → 按最新配置重新注册（停用则不注册） */
|
||||
async reload(jobKey: string): Promise<void> {
|
||||
this.unregister(jobKey);
|
||||
|
||||
const job = await this.jobModel.findOne({ where: { jobKey } });
|
||||
if (!job || Number(job.status) !== 1) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.register(job);
|
||||
}
|
||||
|
||||
/** 立即执行一次入口（手动触发同样按配置重试，但不阻塞调用方） */
|
||||
async runOnce(jobId: number, triggerType: number): Promise<JobRunResult> {
|
||||
return this.execute(jobId, triggerType);
|
||||
}
|
||||
|
||||
/** 校验 cron 表达式合法性 */
|
||||
validateCron(expression: string): boolean {
|
||||
if (!expression || typeof expression !== "string") {
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
new CronTime(expression, TIMEZONE);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/** 计算下次执行时间；表达式非法返回 null */
|
||||
nextRunTime(cronExpression: string): Date | null {
|
||||
if (!this.validateCron(cronExpression)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
try {
|
||||
// luxon 3.x 中 toDate 已移除，使用 toJSDate
|
||||
return new CronTime(cronExpression, TIMEZONE).sendAt().toJSDate();
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** 清理 sys_job_log 中早于 keepDays 天的记录，返回删除条数 */
|
||||
async cleanJobLogs(keepDays: number): Promise<number> {
|
||||
const cutoff = dayjs().subtract(keepDays, "day").toDate();
|
||||
|
||||
return this.jobLogModel.destroy({
|
||||
where: {
|
||||
created_at: { [Op.lt]: cutoff }
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** 按任务配置注册定时器 */
|
||||
private register(job: SysJob): void {
|
||||
this.unregister(job.jobKey);
|
||||
|
||||
const builtin = getBuiltinJob(job.jobKey);
|
||||
if (!builtin) {
|
||||
this.jobLogger.warn(`任务 ${job.jobKey} 未在注册表中实现，跳过调度注册`);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!this.validateCron(job.cronExpression)) {
|
||||
this.jobLogger.warn(`任务 ${job.jobKey} 的 cron 表达式不合法（${job.cronExpression}），跳过调度注册`);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const timer = new CronJob(
|
||||
job.cronExpression,
|
||||
() => {
|
||||
void this.execute(job.id, 1);
|
||||
},
|
||||
null,
|
||||
true,
|
||||
TIMEZONE
|
||||
);
|
||||
this.timers.set(job.jobKey, timer);
|
||||
this.jobLogger.info(`任务 ${job.jobKey} 已注册调度，cron=${job.cronExpression}`);
|
||||
} catch (err) {
|
||||
this.jobLogger.warn(`任务 ${job.jobKey} 定时器创建失败:`, err);
|
||||
}
|
||||
}
|
||||
|
||||
/** 注销定时器与待执行的重试定时器（不清重入锁） */
|
||||
private unregister(jobKey: string): void {
|
||||
const timer = this.timers.get(jobKey);
|
||||
if (timer) {
|
||||
try {
|
||||
timer.stop();
|
||||
} catch (err) {
|
||||
console.warn(`[定时任务] 停止任务 ${jobKey} 定时器失败:`, err);
|
||||
}
|
||||
this.timers.delete(jobKey);
|
||||
}
|
||||
|
||||
const retryTimer = this.retryTimers.get(jobKey);
|
||||
if (retryTimer) {
|
||||
clearTimeout(retryTimer);
|
||||
this.retryTimers.delete(jobKey);
|
||||
}
|
||||
}
|
||||
|
||||
/** 执行内核：重入保护 → 执行尝试 → 失败按配置进入重试链 */
|
||||
private async execute(jobId: number, triggerType: number): Promise<JobRunResult> {
|
||||
const job = await this.jobModel.findByPk(jobId);
|
||||
if (!job) {
|
||||
return { status: 2, message: "任务不存在", costMs: 0, attempt: 0 };
|
||||
}
|
||||
|
||||
const builtin = getBuiltinJob(job.jobKey);
|
||||
if (!builtin) {
|
||||
const now = new Date();
|
||||
await this.writeJobLog({
|
||||
job,
|
||||
triggerType,
|
||||
status: 2,
|
||||
attempt: 1,
|
||||
startedAt: now,
|
||||
finishedAt: now,
|
||||
costMs: 0,
|
||||
message: "任务实现不存在",
|
||||
errorStack: null
|
||||
});
|
||||
await this.writeLastRun(job, { status: 2, message: "任务实现不存在", costMs: 0, attempt: 1 });
|
||||
return { status: 2, message: "任务实现不存在", costMs: 0, attempt: 1 };
|
||||
}
|
||||
|
||||
// 并发保护：上一次尚未结束则本次跳过，且不改写 sys_job.last_*
|
||||
if (this.running.has(job.jobKey)) {
|
||||
const message = "上一次执行尚未结束，本次跳过";
|
||||
const now = new Date();
|
||||
await this.writeJobLog({
|
||||
job,
|
||||
triggerType,
|
||||
status: 3,
|
||||
attempt: 1,
|
||||
startedAt: now,
|
||||
finishedAt: now,
|
||||
costMs: 0,
|
||||
message,
|
||||
errorStack: null
|
||||
});
|
||||
return { status: 3, message, costMs: 0, attempt: 1 };
|
||||
}
|
||||
|
||||
this.running.add(job.jobKey);
|
||||
// 失败且需重试时把锁移交给重试链，本次不释放
|
||||
let handedOverToRetry = false;
|
||||
|
||||
try {
|
||||
const result = await this.runAttempt(job, builtin, triggerType, 1);
|
||||
|
||||
if (result.status !== 1 && Number(job.retryTimes) > 0) {
|
||||
handedOverToRetry = true;
|
||||
this.scheduleRetry(job, 2);
|
||||
} else {
|
||||
await this.writeLastRun(job, result);
|
||||
}
|
||||
|
||||
return { status: result.status, message: result.message, costMs: result.costMs, attempt: result.attempt };
|
||||
} catch (err) {
|
||||
const e = err as any;
|
||||
this.jobLogger.error(`任务 ${job.jobKey} 执行流程异常:`, err);
|
||||
return { status: 2, message: e?.message ? String(e.message) : "执行失败", costMs: 0, attempt: 1 };
|
||||
} finally {
|
||||
if (!handedOverToRetry) {
|
||||
this.running.delete(job.jobKey);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** 重试链：失败重试（trigger_type = 3），重试期间保持重入锁 */
|
||||
private async continueChain(jobId: number, attempt: number, jobKeyHint?: string): Promise<void> {
|
||||
let jobKey = jobKeyHint;
|
||||
let handedOverToRetry = false;
|
||||
|
||||
try {
|
||||
const job = await this.jobModel.findByPk(jobId);
|
||||
if (!job) {
|
||||
return;
|
||||
}
|
||||
jobKey = job.jobKey;
|
||||
|
||||
const builtin = getBuiltinJob(job.jobKey);
|
||||
if (!builtin) {
|
||||
const now = new Date();
|
||||
await this.writeJobLog({
|
||||
job,
|
||||
triggerType: 3,
|
||||
status: 2,
|
||||
attempt,
|
||||
startedAt: now,
|
||||
finishedAt: now,
|
||||
costMs: 0,
|
||||
message: "任务实现不存在",
|
||||
errorStack: null
|
||||
});
|
||||
await this.writeLastRun(job, { status: 2, message: "任务实现不存在", costMs: 0, attempt });
|
||||
return;
|
||||
}
|
||||
|
||||
const result = await this.runAttempt(job, builtin, 3, attempt);
|
||||
|
||||
if (result.status !== 1 && attempt <= Number(job.retryTimes)) {
|
||||
handedOverToRetry = true;
|
||||
this.scheduleRetry(job, attempt + 1);
|
||||
return;
|
||||
}
|
||||
|
||||
// 重试链结束（成功或超出重试次数）才回写任务的 last_run_*
|
||||
await this.writeLastRun(job, result);
|
||||
} catch (err) {
|
||||
this.jobLogger.error(`任务重试执行异常:`, err);
|
||||
} finally {
|
||||
if (!handedOverToRetry && jobKey) {
|
||||
this.running.delete(jobKey);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** 单次尝试：执行 handler 并写入一条执行日志 */
|
||||
private async runAttempt(
|
||||
job: SysJob,
|
||||
builtin: BuiltinJob,
|
||||
triggerType: number,
|
||||
attempt: number
|
||||
): Promise<JobRunResult & { error?: any }> {
|
||||
const startedAt = new Date();
|
||||
const startMs = Date.now();
|
||||
let status = 1;
|
||||
let message = "执行成功";
|
||||
let errorStack: string | null = null;
|
||||
let error: any;
|
||||
|
||||
try {
|
||||
const ctx: JobHandlerContext = {
|
||||
jobKey: job.jobKey,
|
||||
jobName: job.jobName,
|
||||
triggerType,
|
||||
attempt,
|
||||
runtime: this.buildRuntime()
|
||||
};
|
||||
const ret = await builtin.handler(job.params ?? {}, ctx);
|
||||
message = typeof ret === "string" && ret ? ret : "执行成功";
|
||||
} catch (err) {
|
||||
const e = err as any;
|
||||
status = 2;
|
||||
message = e?.message ? String(e.message) : "执行失败";
|
||||
errorStack = (e?.stack ? String(e.stack) : String(e)).slice(0, MAX_ERROR_STACK);
|
||||
error = err;
|
||||
}
|
||||
|
||||
const finishedAt = new Date();
|
||||
const costMs = Date.now() - startMs;
|
||||
|
||||
await this.writeJobLog({ job, triggerType, status, attempt, startedAt, finishedAt, costMs, message, errorStack });
|
||||
|
||||
return { status, message, costMs, attempt, error };
|
||||
}
|
||||
|
||||
/** 写入一条执行日志（写入失败不影响任务执行） */
|
||||
private async writeJobLog(fields: {
|
||||
job: SysJob;
|
||||
triggerType: number;
|
||||
status: number;
|
||||
attempt: number;
|
||||
startedAt: Date;
|
||||
finishedAt: Date;
|
||||
costMs: number;
|
||||
message: string;
|
||||
errorStack: string | null;
|
||||
}): Promise<void> {
|
||||
try {
|
||||
await this.jobLogModel.create({
|
||||
jobId: fields.job.id,
|
||||
jobKey: fields.job.jobKey,
|
||||
jobName: fields.job.jobName,
|
||||
triggerType: fields.triggerType,
|
||||
status: fields.status,
|
||||
attempt: fields.attempt,
|
||||
startedAt: fields.startedAt,
|
||||
finishedAt: fields.finishedAt,
|
||||
costMs: fields.costMs,
|
||||
message: fields.message,
|
||||
errorStack: fields.errorStack
|
||||
} as any);
|
||||
} catch (err) {
|
||||
this.jobLogger.error(`任务 ${fields.job.jobKey} 执行日志写入失败:`, err);
|
||||
}
|
||||
}
|
||||
|
||||
/** 回写任务上次执行信息 */
|
||||
private async writeLastRun(job: SysJob, result: JobRunResult): Promise<void> {
|
||||
await this.jobModel.update(
|
||||
{
|
||||
lastRunAt: new Date(),
|
||||
lastRunStatus: result.status,
|
||||
lastRunCostMs: result.costMs,
|
||||
lastError: result.status === 1 ? null : result.message
|
||||
} as any,
|
||||
{ where: { id: job.id } }
|
||||
);
|
||||
}
|
||||
|
||||
/** 排定一次重试（重试期间保持重入锁） */
|
||||
private scheduleRetry(job: SysJob, attempt: number): void {
|
||||
const intervalSec = Math.max(10, Number(job.retryInterval) || 60);
|
||||
|
||||
const retryTimer = setTimeout(() => {
|
||||
this.retryTimers.delete(job.jobKey);
|
||||
void this.continueChain(job.id, attempt, job.jobKey);
|
||||
}, intervalSec * 1000);
|
||||
|
||||
this.retryTimers.set(job.jobKey, retryTimer);
|
||||
}
|
||||
|
||||
/** 构造注入给任务的运行时能力 */
|
||||
private buildRuntime(): JobRuntime {
|
||||
return {
|
||||
logger: this.jobLogger,
|
||||
// LogService 是请求作用域，单例中改为按需从匿名请求上下文获取
|
||||
cleanOperationLogs: async (keepDays: number) => {
|
||||
const ctx = this.app.createAnonymousContext();
|
||||
const logService = (await ctx.requestContext.getAsync(LogService)) as LogService;
|
||||
return logService.cleanExpiredLogs(keepDays);
|
||||
},
|
||||
cleanJobLogs: (keepDays: number) => this.cleanJobLogs(keepDays)
|
||||
};
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user