Refactor scheduler to use CronExpressionParser
This commit is contained in:
parent
7fde07f107
commit
3fa793bef6
1 changed files with 69 additions and 2 deletions
|
|
@ -1,4 +1,4 @@
|
||||||
import cron, { type ScheduledTask } from "node-cron";
|
import CronExpressionParser from "cron-parser";
|
||||||
import { logger } from "../utils/logger";
|
import { logger } from "../utils/logger";
|
||||||
|
|
||||||
export abstract class Job {
|
export abstract class Job {
|
||||||
|
|
@ -7,6 +7,73 @@ export abstract class Job {
|
||||||
|
|
||||||
type JobConstructor = new () => Job;
|
type JobConstructor = new () => Job;
|
||||||
|
|
||||||
|
class ScheduledTask {
|
||||||
|
private timer: ReturnType<typeof setTimeout> | null = null;
|
||||||
|
private active = true;
|
||||||
|
private running = false;
|
||||||
|
|
||||||
|
constructor(
|
||||||
|
private readonly jobName: string,
|
||||||
|
private readonly cronExpression: string,
|
||||||
|
private readonly run: () => Promise<void>,
|
||||||
|
) {
|
||||||
|
this.scheduleNext();
|
||||||
|
}
|
||||||
|
|
||||||
|
private getDelay(fromDate: Date) {
|
||||||
|
try {
|
||||||
|
const interval = CronExpressionParser.parse(this.cronExpression, {
|
||||||
|
currentDate: fromDate,
|
||||||
|
tz: Intl.DateTimeFormat().resolvedOptions().timeZone,
|
||||||
|
});
|
||||||
|
const nextRun = interval.next().toDate();
|
||||||
|
return Math.max(0, nextRun.getTime() - Date.now());
|
||||||
|
} catch (error) {
|
||||||
|
logger.error(`Failed to parse cron expression for ${this.jobName}:`, error);
|
||||||
|
return 60_000;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private scheduleNext(fromDate = new Date()) {
|
||||||
|
if (!this.active) return;
|
||||||
|
|
||||||
|
const delay = this.getDelay(fromDate);
|
||||||
|
this.timer = setTimeout(() => {
|
||||||
|
void this.tick();
|
||||||
|
}, delay);
|
||||||
|
}
|
||||||
|
|
||||||
|
private async tick() {
|
||||||
|
if (!this.active) return;
|
||||||
|
|
||||||
|
if (this.running) {
|
||||||
|
logger.warn(`Skipping overlapping run for job ${this.jobName}`);
|
||||||
|
this.scheduleNext(new Date());
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
this.running = true;
|
||||||
|
try {
|
||||||
|
await this.run();
|
||||||
|
} finally {
|
||||||
|
this.running = false;
|
||||||
|
this.scheduleNext(new Date());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async stop() {
|
||||||
|
this.active = false;
|
||||||
|
if (this.timer) {
|
||||||
|
clearTimeout(this.timer);
|
||||||
|
this.timer = null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async destroy() {
|
||||||
|
await this.stop();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
class SchedulerClass {
|
class SchedulerClass {
|
||||||
private tasks: ScheduledTask[] = [];
|
private tasks: ScheduledTask[] = [];
|
||||||
|
|
||||||
|
|
@ -18,7 +85,7 @@ class SchedulerClass {
|
||||||
const job = new JobClass();
|
const job = new JobClass();
|
||||||
return {
|
return {
|
||||||
schedule: (cronExpression: string) => {
|
schedule: (cronExpression: string) => {
|
||||||
const task = cron.schedule(cronExpression, async () => {
|
const task = new ScheduledTask(JobClass.name, cronExpression, async () => {
|
||||||
try {
|
try {
|
||||||
await job.run();
|
await job.run();
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue