split worker into a scheduler and worker

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",

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,

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");
 		}
 	};
 

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<Monitor>): Promise<Monitor[] | null> => {
@@ -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<Monitor> }) => {
 		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<Monitor> => {
 		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);

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

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<JobType, number> = {
 	"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<JobType | "heartbeat", NodeJS.Timeout>();
 	private inFlight: Record<JobType, number> = {
 		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<JobType, number> = {
-		check: POLL_MS,
-		"geo-check": POLL_MS,
-		evaluate: POLL_MS,
-		"cleanup-orphaned": POLL_MS,
-		"cleanup-retention": POLL_MS,
-	};
-	private ticking: Partial<Record<JobType, boolean>> = {}; // Whether a loop is mid tick
-	private tickFns = new Map<JobType, () => 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<WorkerMetrics> => {
-		const rows = await this.jobsRepository.findAll();
-		const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS);
-		const now = Date.now();
-		return rows.reduce<WorkerMetrics>(
-			(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 };

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<boolean>;
 	addJob(monitorId: string, monitor: Monitor): Promise<void>;
 	deleteJob(monitor: Monitor): Promise<void>;
 	pauseJob(monitor: Monitor): Promise<void>;
 	resumeJob(monitor: Monitor): Promise<void>;
 	updateJob(monitor: Monitor): Promise<void>;
 	wake(type: JobType): void;
-	shutdown(): Promise<void>;
 	getMetrics(): Promise<WorkerMetrics>;
 	getJobs(pagination: WorkerJobsPagination): Promise<WorkerJobsPage>;
 	flushQueues(): Promise<{ success: boolean }>;
+	shutdown(): Promise<void>;
+}
+
+export interface IQueueWorker extends IJobScheduler {
+	readonly serviceName: string;
+	init(): Promise<boolean>;
 }

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<Record<JobType, boolean>> = {}; // Whether a loop is mid tick
+	protected tickFns = new Map<JobType, () => void>(); // Store tick fns so we can run them immediately on wake
+	protected timers = new Map<JobType | "heartbeat", NodeJS.Timeout>();
+
+	// Per job type poll intervals.  These back off on idle and reset on work or wake()
+	protected pollMs: Record<JobType, number> = {
+		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<WorkerMetrics> => {
+		const rows = await this.jobsRepository.findAll();
+		const workers = await this.queueWorkersRepository.findRecent(WORKER_STALE_MS);
+		const now = Date.now();
+		return rows.reduce<WorkerMetrics>(
+			(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);
+	};
+}