reorg jobqueue to worker

server/src/worker/worker.helper.ts

@@ -0,0 +1,434 @@
+const SERVICE_NAME = "JobQueueHelper";
+import type { Monitor } from "@/domain/monitors/monitor.types.js";
+import { supportsGeoCheck } from "@/domain/monitors/monitor.types.js";
+import { AppError } from "@/utils/AppError.js";
+import { ISettingsService } from "@/domain/app-settings/app-settings.service.js";
+import { ICheckService } from "@/domain/checks/check.service.js";
+import { type IGeoChecksService } from "@/domain/geo-checks/geo-check.service.js";
+import { IncidentService } from "@/domain/incidents/incident.service.js";
+import { INotificationsService } from "@/domain/notifications/notification.service.js";
+import { INetworkService } from "@/service/networkService.js";
+import { IStatusService } from "@/service/statusService.js";
+import { CHECK_TTL_SENTINEL } from "@/domain/checks/check.type.js";
+import { type StatusChangeResult } from "@/types/network.js";
+import { isWindowActive } from "@/utils/maintenanceWindow.js";
+import { IChecksRepository } from "@/domain/checks/check.repository.interface.js";
+import { IGeoChecksRepository } from "@/domain/geo-checks/geo-check.repository.interface.js";
+import { IIncidentsRepository } from "@/domain/incidents/incident.repository.interface.js";
+import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js";
+import { IMonitorStatsRepository } from "@/domain/monitor-stats/monitor-stats.repository.interface.js";
+import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
+import { ITeamsRepository } from "@/domain/teams/team.repository.interface.js";
+import { ILogger } from "@/utils/logger.js";
+import { IBufferService } from "@/service/bufferService.js";
+
+export interface IWorkerHelper {
+	readonly serviceName: string;
+	getHeartbeatJob(): (monitor: Monitor) => Promise<void>;
+	getHeartbeatGeoJob(): (monitor: Monitor) => Promise<void>;
+	getCleanupOrphanedJob(): () => Promise<void>;
+	getCleanupRetentionJob(): () => Promise<void>;
+	isInMaintenanceWindow(monitorId: string, teamId: string): Promise<boolean>;
+}
+
+export interface MonitorActionDecision {
+	shouldCreateIncident: boolean;
+	shouldResolveIncident: boolean;
+	shouldSendNotification: boolean;
+	incidentReason: "status_down" | "threshold_breach" | null;
+	notificationReason: "status_change" | "threshold_breach" | null;
+	thresholdBreaches?: {
+		cpu?: boolean;
+		memory?: boolean;
+		disk?: boolean;
+		temp?: boolean;
+	};
+}
+
+export class QueueHelper implements IWorkerHelper {
+	static SERVICE_NAME = SERVICE_NAME;
+
+	private logger: ILogger;
+	private networkService: INetworkService;
+	private statusService: IStatusService;
+	private notificationsService: INotificationsService;
+	private checkService: ICheckService;
+	private settingsService: ISettingsService;
+	private buffer: IBufferService;
+	private incidentService: IncidentService;
+	private maintenanceWindowsRepository: IMaintenanceWindowsRepository;
+	private monitorsRepository: IMonitorsRepository;
+	private teamsRepository: ITeamsRepository;
+	private monitorStatsRepository: IMonitorStatsRepository;
+	private checksRepository: IChecksRepository;
+	private incidentsRepository: IIncidentsRepository;
+	private geoChecksService: IGeoChecksService;
+	private geoChecksRepository: IGeoChecksRepository;
+
+	constructor(
+		logger: ILogger,
+		networkService: INetworkService,
+		statusService: IStatusService,
+		notificationsService: INotificationsService,
+		checkService: ICheckService,
+		settingsService: ISettingsService,
+		buffer: IBufferService,
+		incidentService: IncidentService,
+		maintenanceWindowsRepository: IMaintenanceWindowsRepository,
+		monitorsRepository: IMonitorsRepository,
+		teamsRepository: ITeamsRepository,
+		monitorStatsRepository: IMonitorStatsRepository,
+		checksRepository: IChecksRepository,
+		incidentsRepository: IIncidentsRepository,
+		geoChecksService: IGeoChecksService,
+		geoChecksRepository: IGeoChecksRepository
+	) {
+		this.logger = logger;
+		this.networkService = networkService;
+		this.statusService = statusService;
+		this.checkService = checkService;
+		this.settingsService = settingsService;
+		this.buffer = buffer;
+		this.notificationsService = notificationsService;
+		this.incidentService = incidentService;
+		this.maintenanceWindowsRepository = maintenanceWindowsRepository;
+		this.monitorsRepository = monitorsRepository;
+		this.teamsRepository = teamsRepository;
+		this.monitorStatsRepository = monitorStatsRepository;
+		this.checksRepository = checksRepository;
+		this.incidentsRepository = incidentsRepository;
+		this.geoChecksService = geoChecksService;
+		this.geoChecksRepository = geoChecksRepository;
+	}
+
+	get serviceName() {
+		return QueueHelper.SERVICE_NAME;
+	}
+
+	getHeartbeatJob = () => {
+		return async (monitor: Monitor) => {
+			try {
+				const monitorId = monitor.id;
+				const teamId = monitor.teamId;
+				if (!monitorId) {
+					throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getMonitorJob" });
+				}
+
+				// Step 1.  Check for maintenance window, if found, skip the check
+
+				const maintenanceWindowActive = await this.isInMaintenanceWindow(monitorId, teamId);
+				if (maintenanceWindowActive) {
+					this.logger.debug({
+						message: `Monitor ${monitorId} is in maintenance window`,
+						service: SERVICE_NAME,
+						method: "getMonitorJob",
+					});
+					if (monitor.status !== "maintenance") {
+						await this.monitorsRepository.updateById(monitorId, teamId, { status: "maintenance" });
+					}
+					return;
+				}
+
+				// Step 2.  Request monitor status
+				const status = await this.networkService.requestStatus(monitor);
+				if (!status) {
+					throw new Error("No network response");
+				}
+
+				// Step 3.  Build check
+				const check = this.checkService.buildCheck(status);
+				if (!check) {
+					this.logger.warn({
+						message: `No check could be built for monitor ${monitorId}`,
+						service: SERVICE_NAME,
+						method: "getMonitorJob",
+						details: { code: status.code, message: status.message },
+					});
+					return;
+				}
+				// Step 4 Add check to buffer
+				this.buffer.addToBuffer(check);
+				// Step 4.  Update monitor status
+				const statusChangeResult = await this.statusService.updateMonitorStatus(status, check);
+
+				// Step 5.  Get decisions
+				const decision = this.evaluateMonitorAction(statusChangeResult);
+
+				// Step 6. Handle notifications (best effort, continue even in event of failure, don't wait)
+				if (decision.shouldSendNotification) {
+					this.notificationsService.handleNotifications(statusChangeResult.monitor, status, decision).catch((error: unknown) => {
+						this.logger.error({
+							message: `Error sending notifications for job ${statusChangeResult.monitor.id}: ${error instanceof Error ? error.message : "Unknown error"}`,
+							service: SERVICE_NAME,
+							method: "getMonitorJob",
+							stack: error instanceof Error ? error.stack : undefined,
+						});
+					});
+				}
+
+				// Step 7. Handle incidents
+				try {
+					await this.incidentService.handleIncident(statusChangeResult.monitor, statusChangeResult.code, decision, status);
+				} catch (error: unknown) {
+					this.logger.warn({
+						message: `Error handling incident for job ${monitor.id}: ${error instanceof Error ? error.message : "Unknown error"}`,
+						service: SERVICE_NAME,
+						method: "getMonitorJob",
+						stack: error instanceof Error ? error.stack : undefined,
+					});
+				}
+			} catch (error: unknown) {
+				this.logger.warn({
+					message: error instanceof Error ? error.message : "Unknown error",
+					service: SERVICE_NAME,
+					method: "getMonitorJob",
+					stack: error instanceof Error ? error.stack : undefined,
+				});
+				throw error;
+			}
+		};
+	};
+
+	getCleanupOrphanedJob = () => {
+		return async () => {
+			try {
+				this.logger.info({
+					message: "Starting cleanup of orphaned data",
+					service: SERVICE_NAME,
+					method: "getCleanupOrphanedJob",
+				});
+
+				// Get all valid team IDs
+				const validTeamIds = await this.teamsRepository.findAllTeamIds();
+				this.logger.debug({
+					message: `Found ${validTeamIds.length} valid teams`,
+					service: SERVICE_NAME,
+					method: "getCleanupOrphanedJob",
+				});
+
+				// Remove orphaned monitors (monitors without a valid team)
+				const deletedMonitorCount = await this.monitorsRepository.deleteByTeamIdsNotIn(validTeamIds);
+				if (deletedMonitorCount > 0) {
+					this.logger.info({
+						message: `Deleted ${deletedMonitorCount} orphaned monitors`,
+						service: SERVICE_NAME,
+						method: "getCleanupOrphanedJob",
+					});
+				}
+
+				// Remove orphaned monitorStats (stats without a valid monitor)
+				const allMonitorIds = await this.monitorsRepository.findAllMonitorIds();
+				this.logger.debug({
+					message: `Found ${allMonitorIds.length} valid monitors`,
+					service: SERVICE_NAME,
+					method: "getCleanupOrphanedJob",
+				});
+
+				const deletedStatsCount = await this.monitorStatsRepository.deleteByMonitorIdsNotIn(allMonitorIds);
+				if (deletedStatsCount > 0) {
+					this.logger.info({
+						message: `Deleted ${deletedStatsCount} orphaned monitor stats`,
+						service: SERVICE_NAME,
+						method: "getCleanupOrphanedJob",
+					});
+				}
+
+				// Remove orphaned checks
+				const deletedChecksCount = await this.checksRepository.deleteByMonitorIdsNotIn(allMonitorIds);
+				if (deletedChecksCount > 0) {
+					this.logger.info({
+						message: `Deleted ${deletedChecksCount} orphaned checks`,
+						service: SERVICE_NAME,
+						method: "getCleanupOrphanedJob",
+					});
+				}
+
+				// Remove orphaned incidents
+				const deletedIncidentsCount = await this.incidentsRepository.deleteByMonitorIdsNotIn(allMonitorIds);
+				if (deletedIncidentsCount > 0) {
+					this.logger.info({
+						message: `Deleted ${deletedIncidentsCount} orphaned incidents`,
+						service: SERVICE_NAME,
+						method: "getCleanupOrphanedJob",
+					});
+				}
+
+				// Remove orphaned geo checks
+				const deletedGeoChecksCount = await this.geoChecksRepository.deleteByMonitorIdsNotIn(allMonitorIds);
+				if (deletedGeoChecksCount > 0) {
+					this.logger.info({
+						message: `Deleted ${deletedGeoChecksCount} orphaned geo checks`,
+						service: SERVICE_NAME,
+						method: "getCleanupOrphanedJob",
+					});
+				}
+
+				this.logger.info({
+					message: "Cleanup of orphaned data completed",
+					service: SERVICE_NAME,
+					method: "getCleanupOrphanedJob",
+				});
+			} catch (error: unknown) {
+				this.logger.warn({
+					message: error instanceof Error ? error.message : "Unknown error",
+					service: SERVICE_NAME,
+					method: "getCleanupOrphanedJob",
+					stack: error instanceof Error ? error.stack : undefined,
+				});
+				throw error;
+			}
+		};
+	};
+
+	getHeartbeatGeoJob = () => {
+		return async (monitor: Monitor) => {
+			try {
+				const monitorId = monitor.id;
+				const teamId = monitor.teamId;
+
+				// Step 1: Validate monitor eligibility
+				if (!monitorId) {
+					throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getHeartbeatGeoJob" });
+				}
+
+				if (!monitor.geoCheckEnabled) {
+					return;
+				}
+				if (!supportsGeoCheck(monitor.type)) {
+					this.logger.debug({
+						message: `Monitor ${monitorId} type does not support geo checks, skipping`,
+						service: SERVICE_NAME,
+						method: "getHeartbeatGeoJob",
+					});
+					return;
+				}
+
+				if (!monitor.geoCheckLocations || monitor.geoCheckLocations.length === 0) {
+					this.logger.warn({
+						message: `No geo check locations configured for monitor ${monitorId}`,
+						service: SERVICE_NAME,
+						method: "getHeartbeatGeoJob",
+					});
+					return;
+				}
+
+				// Step 2: Check for maintenance window
+				const maintenanceWindowActive = await this.isInMaintenanceWindow(monitorId, teamId);
+				if (maintenanceWindowActive) {
+					this.logger.debug({
+						message: `Monitor ${monitorId} is in maintenance window, skipping geo check`,
+						service: SERVICE_NAME,
+						method: "getHeartbeatGeoJob",
+					});
+					return;
+				}
+
+				// Step 3: Build geo check (handles API calls and polling)
+				const geoCheck = await this.geoChecksService.buildGeoCheck(monitor);
+				if (!geoCheck) {
+					this.logger.warn({
+						message: `No geo check could be built for monitor ${monitorId}`,
+						service: SERVICE_NAME,
+						method: "getHeartbeatGeoJob",
+					});
+					return;
+				}
+
+				// Step 4: Add geo check to buffer
+				this.buffer.addGeoCheckToBuffer(geoCheck);
+
+				this.logger.debug({
+					message: `Geo check job executed for monitor ${monitorId}`,
+					service: SERVICE_NAME,
+					method: "getHeartbeatGeoJob",
+				});
+			} catch (error: unknown) {
+				this.logger.error({
+					message: error instanceof Error ? error.message : "Unknown error",
+					service: SERVICE_NAME,
+					method: "getHeartbeatGeoJob",
+					stack: error instanceof Error ? error.stack : undefined,
+				});
+				// Don't throw - geo check failures shouldn't crash the job scheduler
+			}
+		};
+	};
+
+	async isInMaintenanceWindow(monitorId: string, teamId: string) {
+		const maintenanceWindows = await this.maintenanceWindowsRepository.findByMonitorId(monitorId, teamId);
+		const now = new Date();
+		return maintenanceWindows.some((window) => isWindowActive(window, now));
+	}
+
+	getCleanupRetentionJob = () => {
+		return async () => {
+			try {
+				const settings = await this.settingsService.getDBSettings();
+
+				const checkTTL = settings.checkTTL; // Check TTL is in DAYS, not MS
+
+				if (checkTTL === CHECK_TTL_SENTINEL) {
+					this.logger.info({
+						message: `Check TTL is set to unlimited, skipping cleanup`,
+						service: SERVICE_NAME,
+						method: "getCleanupRetentionJob",
+					});
+					return;
+				}
+				const checkTTLInMs = checkTTL * 24 * 60 * 60 * 1000;
+				const cutoffDate = new Date(Date.now() - checkTTLInMs);
+				const deleteCount = await this.checkService.deleteOlderThan(cutoffDate);
+				this.logger.info({
+					message: `Deleted ${deleteCount} checks older than ${cutoffDate.toISOString()}`,
+					service: SERVICE_NAME,
+					method: "getCleanupRetentionJob",
+				});
+			} catch (error: unknown) {
+				this.logger.error({
+					message: error instanceof Error ? error.message : "Unknown error",
+					service: SERVICE_NAME,
+					method: "getCleanupRetentionJob",
+					stack: error instanceof Error ? error.stack : undefined,
+				});
+			}
+		};
+	};
+
+	private evaluateMonitorAction(statusChangeResult: StatusChangeResult): MonitorActionDecision {
+		const { monitor, statusChanged, prevStatus } = statusChangeResult;
+
+		// Initialize result
+		const decision: MonitorActionDecision = {
+			shouldCreateIncident: false,
+			shouldResolveIncident: false,
+			shouldSendNotification: false,
+			incidentReason: null,
+			notificationReason: null,
+		};
+
+		if (!statusChanged) {
+			return decision;
+		}
+
+		if (monitor.status === "down") {
+			// Monitor went down (unreachable)
+			decision.shouldCreateIncident = true;
+			decision.shouldSendNotification = true;
+			decision.incidentReason = "status_down";
+			decision.notificationReason = "status_change";
+		} else if (monitor.status === "breached") {
+			// Hardware monitor exceeded thresholds
+			decision.shouldCreateIncident = true;
+			decision.shouldSendNotification = true;
+			decision.incidentReason = "threshold_breach";
+			decision.notificationReason = "threshold_breach";
+		} else if (monitor.status === "up" && (prevStatus === "down" || prevStatus === "breached")) {
+			// Monitor recovered from down or breached state
+			decision.shouldResolveIncident = true;
+			decision.shouldSendNotification = true;
+			decision.notificationReason = "status_change";
+		}
+
+		return decision;
+	}
+}

server/src/worker/worker.interface.ts

@@ -0,0 +1,66 @@
+import { Monitor } from "@/domain/monitors/monitor.types.js";
+import { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js";
+
+export type WorkerJobFailure = {
+	monitorId: string | number;
+	monitorUrl: string | null;
+	monitorType: string | null;
+	failedAt: number | null;
+	failCount: number;
+	failReason: string | null;
+};
+
+export type WorkerMetrics = {
+	jobs: number;
+	activeJobs: number;
+	failingJobs: number;
+	jobsWithFailures: WorkerJobFailure[];
+	totalRuns: number;
+	totalFailures: number;
+	workers: QueueWorker[];
+};
+
+export type WorkerJobSummary = {
+	monitorId: string | number;
+	monitorUrl: string | null;
+	monitorType: string | null;
+	monitorInterval: number | null;
+	monitorGeoInterval: number | null;
+	monitorActive: boolean | null;
+	active: boolean;
+	lockedBy: string | null;
+	lockedUntil: number | null;
+	lockedAt: number | null;
+	runCount: number;
+	failCount: number;
+	failReason: string | null;
+	lastRunAt: number | null;
+	lastFinishedAt: number | null;
+	lastRunTook: number | null;
+	lastFailedAt: number | null;
+	repeat: number | null;
+};
+
+export type WorkerJobsPagination = {
+	page?: number;
+	rowsPerPage?: number;
+};
+
+export type WorkerJobsPage = {
+	jobs: WorkerJobSummary[];
+	count: number;
+};
+
+export interface IWorker {
+	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>;
+	shutdown(): Promise<void>;
+	getMetrics(): Promise<WorkerMetrics>;
+	getJobs(pagination: WorkerJobsPagination): Promise<WorkerJobsPage>;
+	flushQueues(): Promise<{ success: boolean }>;
+}

server/src/worker/worker.less-simple.ts

@@ -0,0 +1,427 @@
+import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
+import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js";
+import { ILogger } from "@/utils/logger.js";
+import { MongoStore, Scheduler, IJob, AddJobInput } from "less-simple-scheduler";
+import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js";
+import { IWorkerHelper } from "@/worker/worker.helper.js";
+import { type WorkerMetrics, type WorkerJobsPagination, type WorkerJobsPage, IWorker } from "@/worker/worker.interface.js";
+import { type QueueMode } from "@/domain/app-settings/app-settings.type.js";
+import { EnvConfig } from "@/domain/app-settings/app-settings.service.js";
+const SERVICE_NAME = "JobQueue";
+
+// How long after its last heartbeat a worker is still considered alive.  Must be longer than the scheduler's heartbeat
+const WORKER_TTL_MS = 30_000;
+
+// For discriminating job types
+type MonitorJob = Omit<IJob, "data"> & { data: Monitor };
+const MONITOR_TEMPLATES = new Set<string>(["monitor-job", "geo-check-job"]);
+const isMonitorJob = (job: IJob): job is MonitorJob => MONITOR_TEMPLATES.has(job.template);
+
+export class LessSimpleWorker implements IWorker {
+	static SERVICE_NAME = SERVICE_NAME;
+
+	private logger: ILogger;
+	private helper: IWorkerHelper;
+	private monitorsRepository: IMonitorsRepository;
+	private workersRepository: IQueueWorkersRepository;
+	private readonly scheduler: Scheduler;
+	private queueMode: QueueMode;
+	private heartbeatListener: ((workerId: string) => void) | null = null;
+
+	constructor(
+		logger: ILogger,
+		helper: IWorkerHelper,
+		monitorsRepository: IMonitorsRepository,
+		workersRepository: IQueueWorkersRepository,
+		scheduler: Scheduler,
+		queueMode: QueueMode
+	) {
+		this.logger = logger;
+		this.helper = helper;
+		this.monitorsRepository = monitorsRepository;
+		this.workersRepository = workersRepository;
+		this.scheduler = scheduler;
+		this.queueMode = queueMode;
+		this.registerListeners();
+	}
+
+	get serviceName() {
+		return LessSimpleWorker.SERVICE_NAME;
+	}
+
+	private registerListeners = () => {
+		this.scheduler.on("scheduler:start", () => {
+			this.logger.info({
+				message: "Scheduler started",
+				service: SERVICE_NAME,
+			});
+		});
+
+		this.scheduler.on("scheduler:stop", () => {
+			this.logger.info({
+				message: "Scheduler stopped",
+				service: SERVICE_NAME,
+			});
+		});
+
+		this.scheduler.on("scheduler:error", (workerId, error) => {
+			this.logger.error({
+				message: `${workerId} error: ${error instanceof Error ? error.message : String(error)}`,
+				service: SERVICE_NAME,
+				stack: error instanceof Error ? error.stack : undefined,
+			});
+		});
+
+		this.scheduler.on("job:locked", (workerId, job) => {
+			this.logger.debug({
+				message: `${workerId} ${job.id} locked`,
+				service: SERVICE_NAME,
+			});
+		});
+
+		this.scheduler.on("job:abort", (workerId, job, reason) => {
+			this.logger.warn({
+				message: `${workerId} ${job.id} aborted: ${reason}`,
+				service: SERVICE_NAME,
+			});
+		});
+
+		this.scheduler.on("job:attempt", (workerId, job, attempt) => {
+			this.logger.debug({
+				message: `${workerId} ${job.id} attempt ${attempt}`,
+				service: SERVICE_NAME,
+			});
+		});
+
+		this.scheduler.on("job:complete", (workerId, job) => {
+			this.logger.debug({
+				message: `${workerId} ${job.id} completed successfully`,
+				service: SERVICE_NAME,
+			});
+		});
+
+		this.scheduler.on("job:exhausted", (workerId, job, error) => {
+			this.logger.error({
+				message: `${workerId} ${job.id} exhausted all retries: ${error instanceof Error ? error.message : String(error)}`,
+				service: SERVICE_NAME,
+				stack: error instanceof Error ? error.stack : undefined,
+			});
+		});
+
+		this.scheduler.on("job:fail", (workerId, job, error, attempt) => {
+			this.logger.warn({
+				message: `${workerId} ${job.id} failed on attempt ${attempt}: ${error instanceof Error ? error.message : String(error)}`,
+				service: SERVICE_NAME,
+				stack: error instanceof Error ? error.stack : undefined,
+			});
+		});
+
+		this.scheduler.on("job:start", (workerId, job) => {
+			this.logger.debug({
+				message: `${workerId} ${job.id} started`,
+				service: SERVICE_NAME,
+			});
+		});
+	};
+
+	static async create(
+		logger: ILogger,
+		helper: IWorkerHelper,
+		monitorsRepository: IMonitorsRepository,
+		workersRepository: IQueueWorkersRepository,
+		envSettings: EnvConfig,
+		queueMode: QueueMode
+	) {
+		const store = new MongoStore({
+			url: envSettings.dbConnectionString ?? "mongodb://localhost:27017/uptime_db",
+		});
+		const scheduler = new Scheduler(store, {
+			concurrency: envSettings.queueMode === "primary" ? 0 : 50,
+			processEvery: 100,
+		});
+		const instance = new LessSimpleWorker(logger, helper, monitorsRepository, workersRepository, scheduler, queueMode);
+		await instance.init(queueMode);
+
+		return instance;
+	}
+
+	private startHeartbeat = (queueMode: QueueMode) => {
+		const beat = (
+			workerId: string // This is the heartbeat function that registers workers
+		) =>
+			this.workersRepository.upsert(workerId, queueMode).catch((error: unknown) => {
+				this.logger.warn({
+					message: `Worker heartbeat failed: ${error instanceof Error ? error.message : String(error)}`,
+					service: SERVICE_NAME,
+					method: "startHeartbeat",
+				});
+			});
+		beat(this.scheduler.workerId); // Beat immediately to register
+		if (this.heartbeatListener) {
+			this.scheduler.off("scheduler:heartbeat", this.heartbeatListener); // remove old listeners
+		}
+		this.heartbeatListener = beat; /// Keep a reference to the listener
+		this.scheduler.on("scheduler:heartbeat", beat); // beat on every heartbeat
+	};
+
+	init = async (queueMode: QueueMode = "primary") => {
+		try {
+			await this.scheduler.start();
+			this.scheduler.addTemplate("monitor-job", this.helper.getHeartbeatJob());
+			this.scheduler.addTemplate("geo-check-job", this.helper.getHeartbeatGeoJob());
+			this.scheduler.addTemplate("cleanup-orphaned", this.helper.getCleanupOrphanedJob());
+			this.scheduler.addTemplate("cleanup-retention-job", this.helper.getCleanupRetentionJob());
+
+			// Register in the worker registry
+			this.startHeartbeat(queueMode);
+
+			// If this is a worker, return early - primary will populate queue
+			if (queueMode === "worker") {
+				this.logger.info({
+					message: `Worker queue: ${this.scheduler.workerId} initialized`,
+					service: SERVICE_NAME,
+					method: "init",
+				});
+				return true;
+			}
+
+			const monitors = await this.monitorsRepository.findAll();
+			if (!monitors) {
+				return true;
+			}
+			const jobInputs = monitors.flatMap((monitor) => this.buildJobInputs(monitor));
+			if (jobInputs.length > 0) {
+				await this.scheduler.addJobs(jobInputs);
+			}
+
+			this.scheduler.addJob({ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true, upsert: true });
+			this.scheduler.addJob({ id: "cleanup-retention", template: "cleanup-retention-job", active: true, repeat: 24 * 60 * 60 * 1000, upsert: true });
+
+			return true;
+		} catch (error: unknown) {
+			this.logger.error({
+				message: `Failed to initialize LessSimpleQueue: ${error instanceof Error ? error.message : String(error)}`,
+				service: SERVICE_NAME,
+				method: "init",
+			});
+			return false;
+		}
+	};
+
+	// Builds the scheduler inputs for a monitor: always a monitor-job, plus a
+	// geo-check job when the monitor supports and enables geo checks.
+	private buildJobInputs = (monitor: Monitor): AddJobInput[] => {
+		const inputs: AddJobInput[] = [
+			{
+				id: monitor.id,
+				template: "monitor-job",
+				repeat: monitor.interval,
+				active: monitor.isActive,
+				data: monitor,
+				upsert: true,
+				jitter: true,
+			},
+		];
+
+		if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) {
+			inputs.push({
+				id: `${monitor.id}-geo`,
+				template: "geo-check-job",
+				repeat: monitor.geoCheckInterval,
+				active: monitor.isActive,
+				data: monitor,
+				upsert: true,
+				jitter: true,
+			});
+		}
+
+		return inputs;
+	};
+
+	addJob = async (_monitorId: string, monitor: Monitor) => {
+		await this.scheduler.addJobs(this.buildJobInputs(monitor));
+	};
+
+	deleteJob = async (monitor: Monitor) => {
+		this.scheduler.removeJob(monitor.id);
+		const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`);
+		if (geoJob) await this.scheduler.removeJob(`${monitor.id}-geo`);
+	};
+
+	pauseJob = async (monitor: Monitor) => {
+		const result = await this.scheduler.pauseJob(monitor.id);
+		if (result === false) {
+			throw new Error("Failed to pause monitor");
+		}
+		this.scheduler.updateJob(monitor.id, { data: monitor });
+
+		const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`);
+		if (geoJob) {
+			const geoRes = await this.scheduler.pauseJob(`${monitor.id}-geo`);
+			if (geoRes === false) {
+				this.logger.error({
+					message: `Failed to pause geo check job for monitor ${monitor.id}`,
+					service: SERVICE_NAME,
+					method: "pauseJob",
+				});
+			}
+			this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor });
+		}
+
+		this.logger.debug({
+			message: `Paused monitor ${monitor.id}`,
+			service: SERVICE_NAME,
+			method: "pauseJob",
+		});
+	};
+
+	resumeJob = async (monitor: Monitor) => {
+		const result = await this.scheduler.resumeJob(monitor.id);
+		if (result === false) {
+			throw new Error("Failed to resume monitor");
+		}
+		this.scheduler.updateJob(monitor.id, { data: monitor });
+
+		const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`);
+		if (geoJob) {
+			const geoRes = await this.scheduler.resumeJob(`${monitor.id}-geo`);
+			if (geoRes === false) {
+				this.logger.error({
+					message: `Failed to resume geo check job for monitor ${monitor.id}`,
+					service: SERVICE_NAME,
+					method: "resumeJob",
+				});
+			}
+			this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor });
+		}
+		this.logger.debug({
+			message: `Resumed monitor ${monitor.id}`,
+			service: SERVICE_NAME,
+			method: "resumeJob",
+		});
+	};
+
+	private syncGeoJob = async (monitor: Monitor) => {
+		const geoJobId = `${monitor.id}-geo`;
+		const existingGeoJob = await this.scheduler.getJob(geoJobId);
+
+		// If geoChecks have been disabled, or the monitor type doesn't support them, remove
+		if (!monitor.geoCheckEnabled || !supportsGeoCheck(monitor.type)) {
+			if (existingGeoJob) this.scheduler.removeJob(geoJobId);
+			return;
+		}
+
+		// If the job exists, update it
+		if (existingGeoJob) {
+			this.scheduler.updateJob(geoJobId, { repeat: monitor.geoCheckInterval, active: monitor.isActive, data: monitor });
+			return;
+		}
+
+		// Otherwise, create it
+		this.scheduler.addJob({
+			id: geoJobId,
+			template: "geo-check-job",
+			repeat: monitor.geoCheckInterval,
+			active: monitor.isActive,
+			data: monitor,
+			upsert: true,
+		});
+	};
+
+	updateJob = async (monitor: Monitor) => {
+		this.scheduler.updateJob(monitor.id, { repeat: monitor.interval, data: monitor });
+		await this.syncGeoJob(monitor);
+	};
+
+	shutdown = async () => {
+		// Stop beating before we remove this worker, so a heartbeat firing between
+		// the remove and the scheduler stopping can't re-insert it.
+		if (this.heartbeatListener) {
+			this.scheduler.off("scheduler:heartbeat", this.heartbeatListener);
+			this.heartbeatListener = null;
+		}
+		await this.workersRepository.deleteById(this.scheduler.workerId).catch((error: unknown) => {
+			this.logger.warn({
+				message: `Failed to deregister worker on shutdown: ${error instanceof Error ? error.message : String(error)}`,
+				service: SERVICE_NAME,
+				method: "shutdown",
+			});
+		});
+		this.scheduler.stop();
+	};
+
+	getMetrics = async (): Promise<WorkerMetrics> => {
+		const stats = await this.scheduler.getStats();
+
+		const jobsWithFailures = stats.jobsWithFailures.map((failure) => {
+			// `data` is untyped at the store level: for monitor jobs it's the Monitor,
+			// for system jobs (e.g. cleanup) it has no url/type => null
+			const monitor = failure.data as Monitor | undefined;
+			return {
+				monitorId: failure.id,
+				monitorUrl: monitor?.url ?? null,
+				monitorType: monitor?.type ?? null,
+				failedAt: failure.failedAt,
+				failCount: failure.failCount,
+				failReason: failure.failReason,
+			};
+		});
+
+		const workers = await this.workersRepository.findRecent(WORKER_TTL_MS);
+		return {
+			jobs: stats.jobs,
+			activeJobs: stats.activeJobs,
+			failingJobs: stats.failingJobs,
+			totalRuns: stats.totalRuns,
+			totalFailures: stats.totalFailures,
+			jobsWithFailures,
+			workers,
+		};
+	};
+
+	getJobs = async ({ page = 0, rowsPerPage = 0 }: WorkerJobsPagination): Promise<WorkerJobsPage> => {
+		const listOptions = rowsPerPage > 0 ? { skip: Math.max(page, 0) * rowsPerPage, limit: rowsPerPage } : undefined;
+		const [jobs, count] = await Promise.all([this.scheduler.getJobs(listOptions), this.scheduler.countJobs()]);
+		const mapped = jobs.map((job) => {
+			const monitor = isMonitorJob(job) ? job.data : undefined;
+
+			return {
+				monitorId: job.id,
+				monitorUrl: monitor?.url || null,
+				monitorType: monitor?.type || null,
+				monitorInterval: monitor?.interval || null,
+				monitorGeoInterval: monitor?.geoCheckInterval || null,
+				monitorActive: monitor?.isActive ?? null,
+				active: job.active,
+				repeat: job.repeat ?? null,
+				lockedBy: job.lockedBy ?? null,
+				lockedUntil: job.lockedUntil ?? null,
+				lockedAt: job.lockedAt ?? null,
+				runCount: job.runCount ?? 0,
+				failCount: job.failCount ?? 0,
+				failReason: job.lastError ?? null,
+				lastRunAt: job.lastStartedAt ?? null,
+				lastFinishedAt: job.lastFinishedAt ?? null,
+				lastRunTook: job.lockedAt || !job.lastFinishedAt || !job.lastStartedAt ? null : job.lastFinishedAt - job.lastStartedAt,
+				lastFailedAt: job.lastFailedAt ?? null,
+			};
+		});
+
+		return { jobs: mapped, count };
+	};
+
+	flushQueues = async () => {
+		if (this.queueMode !== "primary") {
+			// Don't allow workers to flush, though this isn't currently reachable since workers have no API endpoints
+			this.logger.warn({ message: "Ignoring flush on non-primary worker", service: SERVICE_NAME, method: "flushQueues" });
+			return { success: false };
+		}
+
+		const flushRes = await this.scheduler.flushJobs(); // Less Simple Queue must flush before stopping as it needs DB
+		const stopRes = await this.scheduler.stop(); // Destructive, other workers will abort their jobs
+		const initRes = await this.init();
+		return {
+			success: Boolean(stopRes && flushRes && initRes),
+		};
+	};
+}

server/src/worker/worker.super-simple.ts

@@ -0,0 +1,384 @@
+import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
+import { ILogger } from "@/utils/logger.js";
+import Scheduler from "super-simple-scheduler";
+import { Monitor, MonitorType, supportsGeoCheck } from "@/domain/monitors/monitor.types.js";
+import { hostname } from "node:os";
+import { randomUUID } from "node:crypto";
+import { QueueMode } from "@/domain/app-settings/app-settings.type.js";
+import { IWorker, WorkerJobSummary } from "@/worker/worker.interface.js";
+import { IWorkerHelper } from "@/worker/worker.helper.js";
+
+const SERVICE_NAME = "JobQueue";
+
+export class SuperSimpleQueue implements IWorker {
+	static SERVICE_NAME = SERVICE_NAME;
+
+	private logger: ILogger;
+	private helper: IWorkerHelper;
+	private workerId: string;
+	private monitorsRepository: IMonitorsRepository;
+	private readonly scheduler: Scheduler;
+
+	constructor(logger: ILogger, helper: IWorkerHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) {
+		this.logger = logger;
+		this.helper = helper;
+		this.monitorsRepository = monitorsRepository;
+		this.scheduler = scheduler;
+		this.registerListeners();
+		this.workerId = `${hostname()}:${process.pid}:${randomUUID()}`;
+	}
+
+	get serviceName() {
+		return SuperSimpleQueue.SERVICE_NAME;
+	}
+
+	private registerListeners = () => {
+		this.scheduler.on("scheduler:start", () => {
+			this.logger.info({
+				message: "Scheduler started",
+				service: SERVICE_NAME,
+			});
+		});
+
+		this.scheduler.on("scheduler:stop", () => {
+			this.logger.info({
+				message: "Scheduler stopped",
+				service: SERVICE_NAME,
+			});
+		});
+
+		this.scheduler.on("scheduler:error", (error) => {
+			this.logger.error({
+				message: `Scheduler error: ${error instanceof Error ? error.message : String(error)}`,
+				service: SERVICE_NAME,
+				stack: error instanceof Error ? error.stack : undefined,
+			});
+		});
+
+		this.scheduler.on("job:abort", (job, reason) => {
+			this.logger.warn({
+				message: `${job.id} aborted: ${reason}`,
+				service: SERVICE_NAME,
+			});
+		});
+
+		this.scheduler.on("job:attempt", (job, attempt) => {
+			this.logger.debug({
+				message: `${job.id} attempt ${attempt}`,
+				service: SERVICE_NAME,
+			});
+		});
+
+		this.scheduler.on("job:complete", (job) => {
+			this.logger.debug({
+				message: `${job.id} completed successfully`,
+				service: SERVICE_NAME,
+			});
+		});
+
+		this.scheduler.on("job:exhausted", (job, error) => {
+			this.logger.error({
+				message: `${job.id} exhausted all retries: ${error instanceof Error ? error.message : String(error)}`,
+				service: SERVICE_NAME,
+				stack: error instanceof Error ? error.stack : undefined,
+			});
+		});
+
+		this.scheduler.on("job:fail", (job, error, attempt) => {
+			this.logger.warn({
+				message: `${job.id} failed on attempt ${attempt}: ${error instanceof Error ? error.message : String(error)}`,
+				service: SERVICE_NAME,
+				stack: error instanceof Error ? error.stack : undefined,
+			});
+		});
+
+		this.scheduler.on("job:start", (job) => {
+			this.logger.debug({
+				message: `${job.id} started`,
+				service: SERVICE_NAME,
+			});
+		});
+	};
+
+	static async create(logger: ILogger, helper: IWorkerHelper, monitorsRepository: IMonitorsRepository) {
+		const scheduler = new Scheduler({
+			concurrency: 50,
+			// storeType: "mongo",
+			// storeType: "redis",
+			// dbUri: envSettings.dbConnectionString,
+		});
+		const instance = new SuperSimpleQueue(logger, helper, monitorsRepository, scheduler);
+		await instance.init();
+
+		return instance;
+	}
+
+	init = async () => {
+		try {
+			this.scheduler.start();
+			this.scheduler.addTemplate("monitor-job", this.helper.getHeartbeatJob());
+			this.scheduler.addTemplate("geo-check-job", this.helper.getHeartbeatGeoJob());
+			this.scheduler.addTemplate("cleanup-orphaned", this.helper.getCleanupOrphanedJob());
+			this.scheduler.addTemplate("cleanup-retention-job", this.helper.getCleanupRetentionJob());
+			const monitors = await this.monitorsRepository.findAll();
+			if (!monitors) {
+				return true;
+			}
+			for (const monitor of monitors) {
+				const randomOffset = Math.floor(Math.random() * 100);
+				setTimeout(() => {
+					this.addJob(monitor.id, monitor);
+				}, randomOffset);
+			}
+
+			this.scheduler.addJob({ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true });
+			this.scheduler.addJob({ id: "cleanup-retention", template: "cleanup-retention-job", active: true, repeat: 24 * 60 * 60 * 1000 });
+
+			return true;
+		} catch (error: unknown) {
+			this.logger.error({
+				message: `Failed to initialize SuperSimpleQueue: ${error instanceof Error ? error.message : String(error)}`,
+				service: SERVICE_NAME,
+				method: "init",
+			});
+			return false;
+		}
+	};
+
+	addJob = async (monitorId: string, monitor: Monitor) => {
+		this.scheduler.addJob({
+			id: monitorId,
+			template: "monitor-job",
+			repeat: monitor.interval,
+			active: monitor.isActive,
+			data: monitor,
+		});
+
+		// Return early if we don't need geo checks
+		if (!supportsGeoCheck(monitor.type)) {
+			return;
+		}
+
+		// Add geo check job if enabled for HTTP monitors
+		if (monitor.geoCheckEnabled) {
+			this.scheduler.addJob({
+				id: `${monitorId}-geo`,
+				template: "geo-check-job",
+				repeat: monitor.geoCheckInterval,
+				active: monitor.isActive,
+				data: monitor,
+			});
+		}
+	};
+
+	deleteJob = async (monitor: Monitor) => {
+		this.scheduler.removeJob(monitor.id);
+		const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`);
+		if (geoJob) await this.scheduler.removeJob(`${monitor.id}-geo`);
+	};
+
+	pauseJob = async (monitor: Monitor) => {
+		const result = await this.scheduler.pauseJob(monitor.id);
+		if (result === false) {
+			throw new Error("Failed to pause monitor");
+		}
+		this.scheduler.updateJob(monitor.id, { data: monitor });
+
+		const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`);
+		if (geoJob) {
+			const geoRes = await this.scheduler.pauseJob(`${monitor.id}-geo`);
+			if (geoRes === false) {
+				this.logger.error({
+					message: `Failed to pause geo check job for monitor ${monitor.id}`,
+					service: SERVICE_NAME,
+					method: "pauseJob",
+				});
+			}
+			this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor });
+		}
+
+		this.logger.debug({
+			message: `Paused monitor ${monitor.id}`,
+			service: SERVICE_NAME,
+			method: "pauseJob",
+		});
+	};
+
+	resumeJob = async (monitor: Monitor) => {
+		const result = await this.scheduler.resumeJob(monitor.id);
+		if (result === false) {
+			throw new Error("Failed to resume monitor");
+		}
+		this.scheduler.updateJob(monitor.id, { data: monitor });
+
+		const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`);
+		if (geoJob) {
+			const geoRes = await this.scheduler.resumeJob(`${monitor.id}-geo`);
+			if (geoRes === false) {
+				this.logger.error({
+					message: `Failed to resume geo check job for monitor ${monitor.id}`,
+					service: SERVICE_NAME,
+					method: "resumeJob",
+				});
+			}
+			this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor });
+		}
+		this.logger.debug({
+			message: `Resumed monitor ${monitor.id}`,
+			service: SERVICE_NAME,
+			method: "resumeJob",
+		});
+	};
+
+	private syncGeoJob = async (monitor: Monitor) => {
+		const geoJobId = `${monitor.id}-geo`;
+		const existingGeoJob = await this.scheduler.getJob(geoJobId);
+
+		// If geoChecks have been disabled, or the monitor type doesn't support them, remove
+		if (!monitor.geoCheckEnabled || !supportsGeoCheck(monitor.type)) {
+			if (existingGeoJob) this.scheduler.removeJob(geoJobId);
+			return;
+		}
+
+		// If the job exists, update it
+		if (existingGeoJob) {
+			this.scheduler.updateJob(geoJobId, { repeat: monitor.geoCheckInterval, active: monitor.isActive, data: monitor });
+			return;
+		}
+
+		// Otherwise, create it
+		this.scheduler.addJob({
+			id: geoJobId,
+			template: "geo-check-job",
+			repeat: monitor.geoCheckInterval,
+			active: monitor.isActive,
+			data: monitor,
+		});
+	};
+
+	updateJob = async (monitor: Monitor) => {
+		this.scheduler.updateJob(monitor.id, { repeat: monitor.interval, data: monitor });
+		await this.syncGeoJob(monitor);
+	};
+
+	shutdown = async () => {
+		this.scheduler.stop();
+	};
+
+	getMetrics = async () => {
+		const jobs = await this.scheduler.getJobs();
+		const metrics: {
+			jobs: number;
+			activeJobs: number;
+			failingJobs: number;
+			jobsWithFailures: Array<{
+				monitorId: string | number;
+				monitorUrl: string;
+				monitorType: MonitorType;
+				failedAt: number | null;
+				failCount: number;
+				failReason: string | null;
+			}>;
+			totalRuns: number;
+			totalFailures: number;
+			workers: Array<{
+				workerId: string;
+				mode: QueueMode;
+				lastSeenAt: number;
+			}>;
+		} = jobs.reduce(
+			(acc, job) => {
+				const runCount = job.runCount ?? 0;
+				const failCount = job.failCount ?? 0;
+				const lastFailedAt = job.lastFailedAt ?? 0;
+				const lastRunAt = job.lastRunAt ?? 0;
+				acc.totalRuns += runCount;
+				acc.totalFailures += failCount;
+				acc.jobs++;
+				if (failCount > 0 && lastFailedAt >= lastRunAt) {
+					acc.failingJobs++;
+				}
+
+				if (job.lockedAt) {
+					acc.activeJobs++;
+				}
+
+				if (failCount > 0) {
+					acc.jobsWithFailures.push({
+						monitorId: job.id,
+						monitorUrl: job.data?.url || null,
+						monitorType: job.data?.type || null,
+						failedAt: job.lastFailedAt ?? null,
+						failCount,
+						failReason: job.lastFailReason ?? null,
+					});
+				}
+				return acc;
+			},
+			{
+				jobs: 0,
+				activeJobs: 0,
+				failingJobs: 0,
+				jobsWithFailures: [] as Array<{
+					monitorId: string | number;
+					monitorUrl: string;
+					monitorType: MonitorType;
+					failedAt: number | null;
+					failCount: number;
+					failReason: string | null;
+				}>,
+				totalRuns: 0,
+				totalFailures: 0,
+				workers: [
+					{
+						workerId: this.workerId,
+						mode: "primary",
+						lastSeenAt: Date.now(),
+					},
+				],
+			}
+		);
+		return metrics;
+	};
+
+	getJobs = async ({ page = 0, rowsPerPage = 0 }: { page?: number; rowsPerPage?: number }): Promise<{ jobs: WorkerJobSummary[]; count: number }> => {
+		const jobs = await this.scheduler.getJobs();
+		const mapped = jobs.map((job) => {
+			return {
+				monitorId: job.id,
+				monitorUrl: job.data?.url || null,
+				monitorType: job.data?.type || null,
+				monitorInterval: job.data?.interval || null,
+				monitorGeoInterval: job.data?.geoCheckInterval || null,
+				monitorActive: job.data?.isActive ?? null,
+				active: job.active,
+				repeat: job.repeat ?? null,
+				lockedBy: job.lockedAt ? this.workerId : null,
+				lockedUntil: null,
+				lockedAt: job.lockedAt ?? null,
+				runCount: job.runCount ?? 0,
+				failCount: job.failCount ?? 0,
+				failReason: job.lastFailReason ?? null,
+				lastRunAt: job.lastRunAt ?? null,
+				lastFinishedAt: job.lastFinishedAt ?? null,
+				lastRunTook: job.lockedAt ? null : (job.lastFinishedAt ?? 0) - (job.lastRunAt ?? 0),
+				lastFailedAt: job.lastFailedAt ?? null,
+			};
+		});
+
+		const start = Math.max(page, 0) * rowsPerPage;
+		return {
+			jobs: rowsPerPage > 0 ? mapped.slice(start, start + rowsPerPage) : mapped,
+			count: mapped.length,
+		};
+	};
+
+	flushQueues = async () => {
+		const stopRes = await this.scheduler.stop();
+		const flushRes = await this.scheduler.flushJobs();
+		const initRes = await this.init();
+		return {
+			success: Boolean(stopRes && flushRes && initRes),
+		};
+	};
+}