diff --git a/server/src/api/controllers/queueController.ts b/server/src/api/controllers/queueController.ts index 0d3e2d1d0..0a21761ab 100755 --- a/server/src/api/controllers/queueController.ts +++ b/server/src/api/controllers/queueController.ts @@ -1,4 +1,4 @@ -import { IWorker } from "@/worker/worker.interface.js"; +import { IJobScheduler } from "@/worker/worker.interface.js"; import { getQueueJobsQueryValidation } from "@/api/validation/queueValidation.js"; import { Request, Response, NextFunction } from "express"; @@ -13,10 +13,7 @@ export interface IJobQueueController { class JobQueueController implements IJobQueueController { static SERVICE_NAME = SERVICE_NAME; - private worker: IWorker; - constructor(worker: IWorker) { - this.worker = worker; - } + constructor(private scheduler: IJobScheduler) {} get serviceName() { return JobQueueController.SERVICE_NAME; @@ -24,7 +21,7 @@ class JobQueueController implements IJobQueueController { getMetrics = async (req: Request, res: Response, next: NextFunction) => { try { - const metrics = await this.worker.getMetrics(); + const metrics = await this.scheduler.getMetrics(); res.status(200).json({ success: true, msg: "Queue metrics fetched successfully", @@ -38,7 +35,7 @@ class JobQueueController implements IJobQueueController { getJobs = async (req: Request, res: Response, next: NextFunction) => { try { const pagination = getQueueJobsQueryValidation.parse(req.query); - const { jobs, count } = await this.worker.getJobs(pagination); + const { jobs, count } = await this.scheduler.getJobs(pagination); return res.status(200).json({ success: true, msg: "Queue jobs fetched successfully", @@ -52,8 +49,8 @@ class JobQueueController implements IJobQueueController { getAllMetrics = async (req: Request, res: Response, next: NextFunction) => { try { const pagination = getQueueJobsQueryValidation.parse(req.query); - const { jobs, count } = await this.worker.getJobs(pagination); - const metrics = await this.worker.getMetrics(); + const { jobs, count } = await this.scheduler.getJobs(pagination); + const metrics = await this.scheduler.getMetrics(); return res.status(200).json({ success: true, msg: "Queue metrics fetched successfully", @@ -66,7 +63,7 @@ class JobQueueController implements IJobQueueController { flushQueue = async (req: Request, res: Response, next: NextFunction) => { try { - const result = await this.worker.flushQueues(); + const result = await this.scheduler.flushQueues(); return res.status(200).json({ success: true, msg: "Queue flushed successfully", diff --git a/server/src/config/services.ts b/server/src/config/services.ts index 2f1826241..7bc59f881 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -31,7 +31,7 @@ import { BufferService, IBufferService } from "@/service/bufferService.js"; import { EmailService, IEmailService } from "@/service/emailService.js"; import { GlobalPingService } from "@/service/globalPingService.js"; import { WorkerHelper } from "@/worker/worker.helper.js"; -import { IWorker } from "@/worker/worker.interface.js"; +import { IQueueWorker } from "@/worker/worker.interface.js"; import { INetworkService, NetworkService } from "@/service/networkService.js"; import { IStatusService, StatusService } from "@/service/statusService.js"; import { MonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js"; @@ -117,7 +117,7 @@ export type InitializedServices = { emailService: IEmailService; bufferService: IBufferService; statusService: IStatusService; - worker: IWorker; + worker: IQueueWorker; userService: IUserService; checkService: ICheckService; geoChecksService: IGeoChecksService; @@ -315,7 +315,7 @@ export const initializeServices = async ({ settingsService, logger, jwt, - worker, + scheduler: worker, monitorsRepository, usersRepository, invitesRepository, @@ -334,10 +334,10 @@ export const initializeServices = async ({ monitorsRepository, maintenanceWindowsRepository, jobsRepository, - worker, + scheduler: worker, }); const monitorService = new MonitorService({ - worker, + scheduler: worker, logger, games, monitorsRepository, diff --git a/server/src/domain/maintenance-windows/maintenance-window.service.ts b/server/src/domain/maintenance-windows/maintenance-window.service.ts index b8601321f..5551aef62 100644 --- a/server/src/domain/maintenance-windows/maintenance-window.service.ts +++ b/server/src/domain/maintenance-windows/maintenance-window.service.ts @@ -1,10 +1,10 @@ import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js"; import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; -import { IWorker } from "@/worker/worker.interface.js"; import type { DurationUnit, MaintenanceWindow } from "@/domain/maintenance-windows/maintenance-window.type.js"; import { AppError } from "@/utils/AppError.js"; import { isWindowActive } from "@/utils/maintenanceWindow.js"; +import { IJobScheduler } from "@/worker/worker.interface.js"; const SERVICE_NAME = "maintenanceWindowService"; @@ -43,23 +43,23 @@ export class MaintenanceWindowService implements IMaintenanceWindowService { private monitorsRepository: IMonitorsRepository; private maintenanceWindowsRepository: IMaintenanceWindowsRepository; private jobsRepository: IJobsRepository; - private worker: IWorker; + private scheduler: IJobScheduler; constructor({ monitorsRepository, maintenanceWindowsRepository, jobsRepository, - worker, + scheduler, }: { monitorsRepository: IMonitorsRepository; maintenanceWindowsRepository: IMaintenanceWindowsRepository; jobsRepository: IJobsRepository; - worker: IWorker; + scheduler: IJobScheduler; }) { this.monitorsRepository = monitorsRepository; this.maintenanceWindowsRepository = maintenanceWindowsRepository; this.jobsRepository = jobsRepository; - this.worker = worker; + this.scheduler = scheduler; } get serviceName() { @@ -84,8 +84,8 @@ export class MaintenanceWindowService implements IMaintenanceWindowService { // Rearm jobs to run soon (now + jitter to avoid herding) await this.jobsRepository.markMonitorsDue(toInitializing, now.getTime()); // Wake the (possibly idle, backed-off) loops so the rearmed jobs run promptly instead of waiting up to POLL_MAX_MS - this.worker.wake("check"); - this.worker.wake("geo-check"); + this.scheduler.wake("check"); + this.scheduler.wake("geo-check"); } }; diff --git a/server/src/domain/monitors/monitor.service.ts b/server/src/domain/monitors/monitor.service.ts index 8fce77723..455a1daff 100644 --- a/server/src/domain/monitors/monitor.service.ts +++ b/server/src/domain/monitors/monitor.service.ts @@ -21,8 +21,8 @@ import type { IStatusPagesRepository } from "@/domain/status-pages/status-page-r import demoMonitorsData from "@/utils/demoMonitors.json" with { type: "json" }; import { AppError } from "@/utils/AppError.js"; import type { ImportedMonitor } from "@/api/validation/monitorValidation.js"; -import { IWorker } from "@/worker/worker.interface.js"; import { ILogger } from "@/utils/logger.js"; +import { IJobScheduler } from "@/worker/worker.interface.js"; const SERVICE_NAME = "MonitorService"; type DateRangeKey = "recent" | "day" | "week" | "month" | "all"; @@ -94,7 +94,7 @@ export interface IMonitorService { export class MonitorService implements IMonitorService { static SERVICE_NAME = SERVICE_NAME; - private worker: IWorker; + private scheduler: IJobScheduler; private logger: ILogger; private games: GamesMap; private monitorsRepository: IMonitorsRepository; @@ -105,7 +105,7 @@ export class MonitorService implements IMonitorService { private incidentsRepository: IIncidentsRepository; constructor({ - worker, + scheduler, logger, games, monitorsRepository, @@ -115,7 +115,7 @@ export class MonitorService implements IMonitorService { statusPagesRepository, incidentsRepository, }: { - worker: IWorker; + scheduler: IJobScheduler; logger: ILogger; games: GamesMap; monitorsRepository: IMonitorsRepository; @@ -125,7 +125,7 @@ export class MonitorService implements IMonitorService { statusPagesRepository: IStatusPagesRepository; incidentsRepository: IIncidentsRepository; }) { - this.worker = worker; + this.scheduler = scheduler; this.logger = logger; this.games = games; this.monitorsRepository = monitorsRepository; @@ -171,7 +171,7 @@ export class MonitorService implements IMonitorService { throw new AppError({ message: "Failed to create monitor", status: 500, service: SERVICE_NAME, method: "createMonitor" }); } - this.worker.addJob(monitor.id, monitor); + this.scheduler.addJob(monitor.id, monitor); }; createMonitors = async (monitors: Array): Promise => { @@ -180,7 +180,7 @@ export class MonitorService implements IMonitorService { throw new AppError({ message: "Failed to create monitors", status: 500, service: SERVICE_NAME, method: "createMonitors" }); } - await Promise.all(createdMonitors.map((monitor) => this.worker.addJob(monitor.id, monitor))); + await Promise.all(createdMonitors.map((monitor) => this.scheduler.addJob(monitor.id, monitor))); return createdMonitors; }; @@ -196,7 +196,7 @@ export class MonitorService implements IMonitorService { })); const demoMonitors = await this.monitorsRepository.createMonitors(monitors as unknown as Monitor[]); - await Promise.all(demoMonitors.map((monitor) => this.worker.addJob(monitor.id, monitor))); + await Promise.all(demoMonitors.map((monitor) => this.scheduler.addJob(monitor.id, monitor))); return demoMonitors; }; @@ -428,7 +428,7 @@ export class MonitorService implements IMonitorService { editMonitor = async ({ teamId, monitorId, body }: { teamId: string; monitorId: string; body: Partial }) => { const editedMonitor = await this.monitorsRepository.updateById(monitorId, teamId, body); - await this.worker.updateJob(editedMonitor); + await this.scheduler.updateJob(editedMonitor); return editedMonitor; }; @@ -448,7 +448,7 @@ export class MonitorService implements IMonitorService { // If notifications were updated, we should update the jobs in the queue if (modifiedCount > 0) { const monitors = await this.monitorsRepository.findByIds(monitorIds); - await Promise.all(monitors.map((monitor) => this.worker.updateJob(monitor))); + await Promise.all(monitors.map((monitor) => this.scheduler.updateJob(monitor))); } return modifiedCount; @@ -457,9 +457,9 @@ export class MonitorService implements IMonitorService { pauseMonitor = async ({ teamId, monitorId }: { teamId: string; monitorId: string }): Promise => { const monitor = await this.monitorsRepository.togglePauseById(monitorId, teamId); if (monitor.isActive) { - await this.worker.resumeJob(monitor); + await this.scheduler.resumeJob(monitor); } else { - await this.worker.pauseJob(monitor); + await this.scheduler.pauseJob(monitor); } return monitor; }; @@ -478,9 +478,9 @@ export class MonitorService implements IMonitorService { const results = await Promise.allSettled( monitors.map(async (monitor) => { if (monitor.isActive) { - await this.worker.resumeJob(monitor); + await this.scheduler.resumeJob(monitor); } else { - await this.worker.pauseJob(monitor); + await this.scheduler.pauseJob(monitor); } }) ); @@ -541,7 +541,7 @@ export class MonitorService implements IMonitorService { }); }); - await this.worker.deleteJob(monitor); + await this.scheduler.deleteJob(monitor); return monitor; }; @@ -550,7 +550,7 @@ export class MonitorService implements IMonitorService { await Promise.all( monitors.map(async (monitor) => { try { - await this.worker.deleteJob(monitor); + await this.scheduler.deleteJob(monitor); await this.checksRepository.deleteByMonitorId(monitor.id); await this.geoChecksRepository.deleteByMonitorId(monitor.id); await this.statusPagesRepository.removeMonitorFromStatusPages(monitor.id); diff --git a/server/src/domain/users/user.service.ts b/server/src/domain/users/user.service.ts index 33d95f166..7bdb9677f 100644 --- a/server/src/domain/users/user.service.ts +++ b/server/src/domain/users/user.service.ts @@ -11,9 +11,9 @@ import { AppError } from "@/utils/AppError.js"; import { IEmailService } from "@/service/emailService.js"; import { EnvConfig, ISettingsService } from "@/domain/app-settings/app-settings.service.js"; import { ILogger } from "@/utils/logger.js"; +import { IJobScheduler } from "@/worker/worker.interface.js"; import jwt from "jsonwebtoken"; import crypto from "crypto"; -import { IWorker } from "@/worker/worker.interface.js"; type CryptoType = typeof crypto; type JwtType = typeof jwt; const SERVICE_NAME = "userService"; @@ -53,7 +53,7 @@ export class UserService implements IUserService { private settingsService: ISettingsService; private logger: ILogger; private jwt: JwtType; - private worker: IWorker; + private scheduler: IJobScheduler; private crypto: CryptoType; private monitorsRepository: IMonitorsRepository; private usersRepository: IUsersRepository; @@ -68,7 +68,7 @@ export class UserService implements IUserService { settingsService, logger, jwt, - worker, + scheduler, monitorsRepository, usersRepository, invitesRepository, @@ -81,7 +81,7 @@ export class UserService implements IUserService { settingsService: ISettingsService; logger: ILogger; jwt: JwtType; - worker: IWorker; + scheduler: IJobScheduler; monitorsRepository: IMonitorsRepository; usersRepository: IUsersRepository; invitesRepository: IInvitesRepository; @@ -93,7 +93,7 @@ export class UserService implements IUserService { this.settingsService = settingsService; this.logger = logger; this.jwt = jwt; - this.worker = worker; + this.scheduler = scheduler; this.crypto = crypto; this.monitorsRepository = monitorsRepository; this.usersRepository = usersRepository; @@ -330,7 +330,7 @@ export class UserService implements IUserService { if (roles.includes("superadmin")) { const monitors = await this.monitorsRepository.findByTeamId(teamId, {}); if (monitors) { - await Promise.all(monitors.map((monitor) => this.worker.deleteJob(monitor))); + await Promise.all(monitors.map((monitor) => this.scheduler.deleteJob(monitor))); } } // 6. Delete the user by id diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index 5e1a74bb4..ec7d1cf94 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -1,5 +1,5 @@ -import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; -import { IWorker, WorkerJobsPagination, WorkerJobSummary, WorkerMetrics } from "@/worker/worker.interface.js"; +import { supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; +import { IQueueWorker } from "@/worker/worker.interface.js"; import { type Job, type JobSeed, type JobType, jobId, LOCK_MS } from "@/domain/jobs/job.type.js"; import { ILogger } from "@/utils/logger.js"; import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; @@ -14,6 +14,7 @@ import { IWorkerHelper } from "@/worker/worker.helper.js"; import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js"; import { WORKER_TTL_SECONDS } from "@/domain/queue-workers/queue-worker.model.js"; import { QueueMode } from "@/domain/app-settings/app-settings.type.js"; +import { JobScheduler } from "@/worker/worker.job-scheduler.js"; const SERVICE_NAME = "JobQueue"; const POLL_MS = 250; // base poll interval while a loop is actively claiming work const POLL_MAX_MS = 5000; // back off poll interval to 5s if no jobs @@ -29,10 +30,8 @@ const CONCURRENCY: Record = { "cleanup-retention": 1, }; -export class DBQueueWorker implements IWorker { +export class DBQueueWorker extends JobScheduler implements IQueueWorker { static SERVICE_NAME = SERVICE_NAME; - private stopped = false; - private timers = new Map(); private inFlight: Record = { check: 0, "geo-check": 0, @@ -41,33 +40,24 @@ export class DBQueueWorker implements IWorker { "cleanup-retention": 0, }; - // Per job type poll intervals. These back off on idle and reset on work or wake() - private pollMs: Record = { - check: POLL_MS, - "geo-check": POLL_MS, - evaluate: POLL_MS, - "cleanup-orphaned": POLL_MS, - "cleanup-retention": POLL_MS, - }; - private ticking: Partial> = {}; // Whether a loop is mid tick - private tickFns = new Map void>(); // Store tick fns so we can run them immediately on wake - constructor( private logger: ILogger, - private jobsRepository: IJobsRepository, + jobsRepository: IJobsRepository, // ← no modifier: forwarded to super private monitorsRepository: IMonitorsRepository, private checksRepository: IChecksRepository, - private checkService: ICheckService, // toStatusResponse + toLastEvaluatedAt + private checkService: ICheckService, private checkProducer: ICheckProducer, private checkEvaluator: ICheckEvaluator, private geoCheckPipeline: ICheckPipeline, private dispatcher: IReactorDispatcher, private helper: IWorkerHelper, - private queueWorkersRepository: IQueueWorkersRepository, + queueWorkersRepository: IQueueWorkersRepository, // ← no modifier: forwarded private queueMode: QueueMode, private queuePrimaryProcesses: boolean, - private readonly workerId: string // same id MongoJobsRepository stamps into lockedBy - ) {} + workerId: string + ) { + super(jobsRepository, queueWorkersRepository, workerId); + } get serviceName() { return DBQueueWorker.SERVICE_NAME; @@ -109,23 +99,6 @@ export class DBQueueWorker implements IWorker { return instance; } - // Helpers, builds job seeds. immediate=true runs on the next tick; otherwise jitter spreads the herd. - private toCheckJob = (monitor: Monitor, now: number, immediate = false): JobSeed => ({ - id: jobId("check", monitor.id), - type: "check", - refId: monitor.id, - isActive: monitor.isActive, - nextScheduledAt: immediate ? now : now + Math.floor(Math.random() * monitor.interval), // Avoid herd - intervalMs: monitor.interval, - }); - - private toGeoCheckJob = (monitor: Monitor, now: number, immediate = false): JobSeed => ({ - ...this.toCheckJob(monitor, now, immediate), - id: jobId("geo-check", monitor.id), - type: "geo-check", - intervalMs: monitor.geoCheckInterval!, - }); - private toCleanupJob = (type: "cleanup-orphaned" | "cleanup-retention", now: number): JobSeed => ({ id: jobId(type, null), type, @@ -135,20 +108,6 @@ export class DBQueueWorker implements IWorker { intervalMs: CLEANUP_INTERVAL_MS, }); - private toSummary = (job: Job): WorkerJobSummary => ({ - monitorId: job.refId ?? job.id, - monitorType: job.type, - monitorInterval: job.intervalMs, - monitorActive: job.isActive, - lockedBy: job.lockedBy, - lockedUntil: job.lockedUntil, - nextScheduledAt: job.nextScheduledAt, - runCount: job.runCount, - failCount: job.failCount, - failReason: job.lastFailReason, - lastFinishedAt: job.lastFinishedAt, - }); - // ******************** // Stage 1: This can eventually be offloaded to a separate service // ******************** @@ -276,15 +235,6 @@ export class DBQueueWorker implements IWorker { // Wake a worker loop after adding a job that's due immediately. // Otherwise the loop could be delayed by POLL_MAX_MS if it is already fully backed off - wake = (type: JobType) => { - this.pollMs[type] = POLL_MS; - if (this.stopped || this.ticking[type]) return; // If it's not running or mid tick, do nothing - const tick = this.tickFns.get(type); - if (!tick) return; // Try to get a handle on the tick fn, if we can't we're done - const pending = this.timers.get(type); - if (pending) clearTimeout(pending); // Clear out scheduled tick if any - this.timers.set(type, setTimeout(tick, 0)); // Set immediate tick - }; // Seed queue private reconcile = async () => { @@ -332,85 +282,7 @@ export class DBQueueWorker implements IWorker { return true; }; - addJob = async (_monitorId: string, monitor: Monitor) => { - const now = Date.now(); - await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now, true)); - this.wake("check"); - if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { - await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, now, true)); - this.wake("geo-check"); - } - }; - deleteJob = async (monitor: Monitor) => { - await this.jobsRepository.deleteById(monitor.id); - }; - pauseJob = async (monitor: Monitor) => { - await this.jobsRepository.setActiveById(monitor.id, false); - }; - resumeJob = async (monitor: Monitor) => { - await this.jobsRepository.setActiveById(monitor.id, true); // a paused job's nextScheduledAt is now in the past, so it's due immediately - this.wake("check"); - if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) this.wake("geo-check"); - }; - updateJob = async (monitor: Monitor) => { - await this.jobsRepository.updateScheduleById(monitor.id, "check", monitor.interval); - if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { - await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, Date.now(), true)); // a newly enabled geo check should run right away - this.wake("geo-check"); - } else { - await this.jobsRepository.deleteByIdAndType(monitor.id, "geo-check"); - } - }; - shutdown = async () => { - this.stopped = true; - // Stop all timers - for (const timer of this.timers.values()) { - clearTimeout(timer); - } - this.timers.clear(); // Clear the map out - // Remove workers from registry - await this.queueWorkersRepository.deleteById(this.workerId); - }; - getMetrics = async (): Promise => { - const rows = await this.jobsRepository.findAll(); - const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS); - const now = Date.now(); - return rows.reduce( - (acc, job) => { - acc.jobs++; - acc.totalRuns += job.runCount; - acc.totalFailures += job.failCount; - if (job.lockedBy !== null && job.lockedUntil !== null && job.lockedUntil > now) acc.activeJobs++; - if (job.failCount > 0) { - acc.failingJobs++; - acc.jobsWithFailures.push({ - monitorId: job.refId ?? job.id, - monitorUrl: null, - monitorType: job.type, - failedAt: job.lastFinishedAt, - failCount: job.failCount, - failReason: job.lastFailReason, - }); - } - return acc; - }, - { - jobs: 0, - activeJobs: 0, - failingJobs: 0, - jobsWithFailures: [], - totalRuns: 0, - totalFailures: 0, - workers, - } - ); - }; - - getJobs = async (pagination: WorkerJobsPagination) => { - const { jobs, count } = await this.jobsRepository.findPage(pagination); - return { jobs: jobs.map(this.toSummary), count }; - }; - flushQueues = async () => { + override flushQueues = async () => { await this.shutdown(); const ok = await this.init(); return { success: ok }; diff --git a/server/src/worker/worker.interface.ts b/server/src/worker/worker.interface.ts index 6dd360e50..522ed3204 100644 --- a/server/src/worker/worker.interface.ts +++ b/server/src/worker/worker.interface.ts @@ -56,17 +56,21 @@ export type WorkerJobsPage = { count: number; }; -export interface IWorker { +export interface IJobScheduler { readonly serviceName: string; - init(): Promise; addJob(monitorId: string, monitor: Monitor): Promise; deleteJob(monitor: Monitor): Promise; pauseJob(monitor: Monitor): Promise; resumeJob(monitor: Monitor): Promise; updateJob(monitor: Monitor): Promise; wake(type: JobType): void; - shutdown(): Promise; getMetrics(): Promise; getJobs(pagination: WorkerJobsPagination): Promise; flushQueues(): Promise<{ success: boolean }>; + shutdown(): Promise; +} + +export interface IQueueWorker extends IJobScheduler { + readonly serviceName: string; + init(): Promise; } diff --git a/server/src/worker/worker.job-scheduler.ts b/server/src/worker/worker.job-scheduler.ts new file mode 100644 index 000000000..4e05e4c38 --- /dev/null +++ b/server/src/worker/worker.job-scheduler.ts @@ -0,0 +1,167 @@ +import { IJobScheduler } from "@/worker/worker.interface.js"; +import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; +import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; +import { type Job, type JobSeed, type JobType, jobId } from "@/domain/jobs/job.type.js"; +import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js"; +import { WorkerJobsPagination, WorkerJobSummary, WorkerMetrics } from "@/worker/worker.interface.js"; +import { WORKER_TTL_SECONDS } from "@/domain/queue-workers/queue-worker.model.js"; + +const POLL_MS = 250; // base poll interval while a loop is actively claiming work +const SERVICE_NAME = "JobScheduler"; +const WORKER_STALE_MS = WORKER_TTL_SECONDS * 1000; // a worker counts as alive if seen within this window + +export class JobScheduler implements IJobScheduler { + static SERVICE_NAME = SERVICE_NAME; + constructor( + protected jobsRepository: IJobsRepository, + protected queueWorkersRepository: IQueueWorkersRepository, + protected readonly workerId: string // same id MongoJobsRepository stamps into lockedBy + ) {} + + get serviceName() { + return JobScheduler.SERVICE_NAME; + } + + protected stopped = false; + protected ticking: Partial> = {}; // Whether a loop is mid tick + protected tickFns = new Map void>(); // Store tick fns so we can run them immediately on wake + protected timers = new Map(); + + // Per job type poll intervals. These back off on idle and reset on work or wake() + protected pollMs: Record = { + check: POLL_MS, + "geo-check": POLL_MS, + evaluate: POLL_MS, + "cleanup-orphaned": POLL_MS, + "cleanup-retention": POLL_MS, + }; + // Helpers, builds job seeds. immediate=true runs on the next tick; otherwise jitter spreads the herd. + protected toCheckJob = (monitor: Monitor, now: number, immediate = false): JobSeed => ({ + id: jobId("check", monitor.id), + type: "check", + refId: monitor.id, + isActive: monitor.isActive, + nextScheduledAt: immediate ? now : now + Math.floor(Math.random() * monitor.interval), // Avoid herd + intervalMs: monitor.interval, + }); + + protected toGeoCheckJob = (monitor: Monitor, now: number, immediate = false): JobSeed => ({ + ...this.toCheckJob(monitor, now, immediate), + id: jobId("geo-check", monitor.id), + type: "geo-check", + intervalMs: monitor.geoCheckInterval!, + }); + + private toSummary = (job: Job): WorkerJobSummary => ({ + monitorId: job.refId ?? job.id, + monitorType: job.type, + monitorInterval: job.intervalMs, + monitorActive: job.isActive, + lockedBy: job.lockedBy, + lockedUntil: job.lockedUntil, + nextScheduledAt: job.nextScheduledAt, + runCount: job.runCount, + failCount: job.failCount, + failReason: job.lastFailReason, + lastFinishedAt: job.lastFinishedAt, + }); + wake = (type: JobType) => { + this.pollMs[type] = POLL_MS; + if (this.stopped || this.ticking[type]) return; // If it's not running or mid tick, do nothing + const tick = this.tickFns.get(type); + if (!tick) return; // Try to get a handle on the tick fn, if we can't we're done + const pending = this.timers.get(type); + if (pending) clearTimeout(pending); // Clear out scheduled tick if any + this.timers.set(type, setTimeout(tick, 0)); // Set immediate tick + }; + + addJob = async (_monitorId: string, monitor: Monitor) => { + const now = Date.now(); + await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now, true)); + this.wake("check"); + if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { + await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, now, true)); + this.wake("geo-check"); + } + }; + + deleteJob = async (monitor: Monitor) => { + await this.jobsRepository.deleteById(monitor.id); + }; + + pauseJob = async (monitor: Monitor) => { + await this.jobsRepository.setActiveById(monitor.id, false); + }; + + resumeJob = async (monitor: Monitor) => { + await this.jobsRepository.setActiveById(monitor.id, true); // a paused job's nextScheduledAt is now in the past, so it's due immediately + this.wake("check"); + if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) this.wake("geo-check"); + }; + + updateJob = async (monitor: Monitor) => { + await this.jobsRepository.updateScheduleById(monitor.id, "check", monitor.interval); + if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) { + await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, Date.now(), true)); // a newly enabled geo check should run right away + this.wake("geo-check"); + } else { + await this.jobsRepository.deleteByIdAndType(monitor.id, "geo-check"); + } + }; + + getMetrics = async (): Promise => { + const rows = await this.jobsRepository.findAll(); + const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS); + const now = Date.now(); + return rows.reduce( + (acc, job) => { + acc.jobs++; + acc.totalRuns += job.runCount; + acc.totalFailures += job.failCount; + if (job.lockedBy !== null && job.lockedUntil !== null && job.lockedUntil > now) acc.activeJobs++; + if (job.failCount > 0) { + acc.failingJobs++; + acc.jobsWithFailures.push({ + monitorId: job.refId ?? job.id, + monitorUrl: null, + monitorType: job.type, + failedAt: job.lastFinishedAt, + failCount: job.failCount, + failReason: job.lastFailReason, + }); + } + return acc; + }, + { + jobs: 0, + activeJobs: 0, + failingJobs: 0, + jobsWithFailures: [], + totalRuns: 0, + totalFailures: 0, + workers, + } + ); + }; + + getJobs = async (pagination: WorkerJobsPagination) => { + const { jobs, count } = await this.jobsRepository.findPage(pagination); + return { jobs: jobs.map(this.toSummary), count }; + }; + + flushQueues = async () => { + await this.shutdown(); + return { success: true }; + }; + + shutdown = async () => { + this.stopped = true; + // Stop all timers + for (const timer of this.timers.values()) { + clearTimeout(timer); + } + this.timers.clear(); // Clear the map out + // Remove workers from registry + await this.queueWorkersRepository.deleteById(this.workerId); + }; +}