smart_grid_iot/server/src/service/job_scheduler.service.ts

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 { NoticeRuleService } from "./notice_rule.service";
import { IccLineSyncService } from "./icc_line_sync.service";
import { IccCustomerSyncService } from "./icc_customer_sync.service";
import { IccGatewaySyncService } from "./icc_gateway_sync.service";
import { IccAnalogSyncService } from "./icc_analog_sync.service";
import { IccEventSubscribeService } from "./icc_event_subscribe.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;
		}
	}

	/** 计算未来 count 次执行时间（配置界面预览用）；表达式非法返回空数组 */
	nextRunTimes(cronExpression: string, count: number): Date[] {
		if (!this.validateCron(cronExpression)) {
			return [];
		}

		try {
			return new CronTime(cronExpression, TIMEZONE)
				.sendAt(count)
				.map(date => date.toJSDate());
		} catch {
			return [];
		}
	}

	/** 清理 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),
			// NoticeRuleService 同样是请求作用域，按需从匿名请求上下文获取
			runNoticeRules: async () => {
				const ctx = this.app.createAnonymousContext();
				const noticeRuleService = (await ctx.requestContext.getAsync(NoticeRuleService)) as NoticeRuleService;
				return noticeRuleService.executeScan();
			},
			syncIccLines: async () => {
				const ctx = this.app.createAnonymousContext();
				const iccService = (await ctx.requestContext.getAsync(IccLineSyncService)) as IccLineSyncService;
				const result = await iccService.syncFromIcc();
				return `拉取 ${result.pulled} 条，新增 ${result.inserted}，更新 ${result.updated}，跳过 ${result.skipped}`;
			},
			syncIccCustomers: async () => {
				const ctx = this.app.createAnonymousContext();
				const iccCustomerService = (await ctx.requestContext.getAsync(IccCustomerSyncService)) as IccCustomerSyncService;
				const result = await iccCustomerService.syncFromIcc();
				return `拉取 ${result.pulled} 条，新增 ${result.inserted}，更新 ${result.updated}，跳过 ${result.skipped}`;
			},
			syncIccGateways: async () => {
				const ctx = this.app.createAnonymousContext();
				const iccGatewayService = (await ctx.requestContext.getAsync(IccGatewaySyncService)) as IccGatewaySyncService;
				const result = await iccGatewayService.syncFromIcc();
				return `拉取 ${result.pulled} 条，新增 ${result.inserted}，更新 ${result.updated}，跳过 ${result.skipped}`;
			},
			syncIccAnalogs: async () => {
				const ctx = this.app.createAnonymousContext();
				const iccAnalogService = (await ctx.requestContext.getAsync(IccAnalogSyncService)) as IccAnalogSyncService;
				const result = await iccAnalogService.syncFromIcc();
				return `拉取 ${result.pulled} 条，新增 ${result.inserted}，更新 ${result.updated}，跳过 ${result.skipped}`;
			},
			ensureIccEventSubscribe: async () => {
				const ctx = this.app.createAnonymousContext();
				const iccEventService = (await ctx.requestContext.getAsync(IccEventSubscribeService)) as IccEventSubscribeService;
				const result = await iccEventService.ensureSubscribe();
				return result.success ? `ICC 事件订阅成功（code=${result.code ?? "0"}）` : `ICC 事件订阅失败：${result.errMsg ?? ""}`;
			}
		};
	}
}