115 lines
4.8 KiB
JavaScript
115 lines
4.8 KiB
JavaScript
"use strict";
|
|
var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
|
|
if (k2 === undefined) k2 = k;
|
|
var desc = Object.getOwnPropertyDescriptor(m, k);
|
|
if (!desc || ("get" in desc ? !m.__esModule : desc.writable || desc.configurable)) {
|
|
desc = { enumerable: true, get: function() { return m[k]; } };
|
|
}
|
|
Object.defineProperty(o, k2, desc);
|
|
}) : (function(o, m, k, k2) {
|
|
if (k2 === undefined) k2 = k;
|
|
o[k2] = m[k];
|
|
}));
|
|
var __setModuleDefault = (this && this.__setModuleDefault) || (Object.create ? (function(o, v) {
|
|
Object.defineProperty(o, "default", { enumerable: true, value: v });
|
|
}) : function(o, v) {
|
|
o["default"] = v;
|
|
});
|
|
var __importStar = (this && this.__importStar) || (function () {
|
|
var ownKeys = function(o) {
|
|
ownKeys = Object.getOwnPropertyNames || function (o) {
|
|
var ar = [];
|
|
for (var k in o) if (Object.prototype.hasOwnProperty.call(o, k)) ar[ar.length] = k;
|
|
return ar;
|
|
};
|
|
return ownKeys(o);
|
|
};
|
|
return function (mod) {
|
|
if (mod && mod.__esModule) return mod;
|
|
var result = {};
|
|
if (mod != null) for (var k = ownKeys(mod), i = 0; i < k.length; i++) if (k[i] !== "default") __createBinding(result, mod, k[i]);
|
|
__setModuleDefault(result, mod);
|
|
return result;
|
|
};
|
|
})();
|
|
Object.defineProperty(exports, "__esModule", { value: true });
|
|
exports.SchedulingService = void 0;
|
|
/**
|
|
* SchedulingService — manages cron jobs for scheduled explorations.
|
|
* On startup, loads all enabled schedules and registers cron tasks.
|
|
* When a schedule fires, it enqueues an exploration job via IJobQueue.
|
|
*/
|
|
const cron = __importStar(require("node-cron"));
|
|
const ExplorationWorker_1 = require("../../../jobs/workers/ExplorationWorker");
|
|
const ScheduleFired_1 = require("../domain/events/ScheduleFired");
|
|
const UniqueId_1 = require("../../../shared/domain/UniqueId");
|
|
class SchedulingService {
|
|
constructor(scheduleRepo, jobQueue, eventBus, logger) {
|
|
this.scheduleRepo = scheduleRepo;
|
|
this.jobQueue = jobQueue;
|
|
this.eventBus = eventBus;
|
|
this.logger = logger;
|
|
this.jobs = new Map();
|
|
}
|
|
async start() {
|
|
const schedules = await this.scheduleRepo.findAll(true);
|
|
for (const schedule of schedules) {
|
|
this.registerCron(schedule);
|
|
}
|
|
this.logger.info({ count: schedules.length }, 'SchedulingService started');
|
|
}
|
|
stop() {
|
|
for (const [id, task] of this.jobs) {
|
|
task.stop();
|
|
this.logger.debug({ scheduleId: id }, 'Cron job stopped');
|
|
}
|
|
this.jobs.clear();
|
|
this.logger.info('SchedulingService stopped');
|
|
}
|
|
registerCron(schedule) {
|
|
this.unregisterCron(schedule.id.toString());
|
|
if (!schedule.enabled)
|
|
return;
|
|
if (!cron.validate(schedule.cronExpression.value)) {
|
|
this.logger.warn({ scheduleId: schedule.id.toString(), cron: schedule.cronExpression.value }, 'Invalid cron, skipping');
|
|
return;
|
|
}
|
|
const task = cron.schedule(schedule.cronExpression.value, () => {
|
|
void this.fire(schedule.id.toString());
|
|
});
|
|
this.jobs.set(schedule.id.toString(), task);
|
|
this.logger.info({ scheduleId: schedule.id.toString(), cron: schedule.cronExpression.value }, 'Cron job registered');
|
|
}
|
|
unregisterCron(scheduleId) {
|
|
const existing = this.jobs.get(scheduleId);
|
|
if (existing) {
|
|
existing.stop();
|
|
this.jobs.delete(scheduleId);
|
|
}
|
|
}
|
|
async fire(scheduleId) {
|
|
const schedule = await this.scheduleRepo.findById(UniqueId_1.UniqueId.from(scheduleId));
|
|
if (!schedule || !schedule.enabled)
|
|
return;
|
|
this.logger.info({ scheduleId, url: schedule.url }, 'Firing scheduled exploration');
|
|
const payload = {
|
|
sessionId: UniqueId_1.UniqueId.create().toString(),
|
|
url: schedule.url,
|
|
seed: Math.floor(Math.random() * 0x7fffffff),
|
|
maxStates: schedule.config['maxStates'] ?? 50,
|
|
config: schedule.config,
|
|
};
|
|
try {
|
|
const jobId = await this.jobQueue.enqueue(ExplorationWorker_1.EXPLORATION_JOB_TYPE, payload);
|
|
schedule.markFired(Date.now());
|
|
await this.scheduleRepo.update(schedule);
|
|
await this.eventBus.publish(new ScheduleFired_1.ScheduleFired(scheduleId, { scheduleId, url: schedule.url, jobId }));
|
|
this.logger.info({ scheduleId, jobId, url: schedule.url }, 'Scheduled exploration enqueued');
|
|
}
|
|
catch (err) {
|
|
this.logger.error({ scheduleId, err: err instanceof Error ? err.message : String(err) }, 'Failed to enqueue scheduled exploration');
|
|
}
|
|
}
|
|
}
|
|
exports.SchedulingService = SchedulingService;
|