Merge remote-tracking branch 'upstream/develop' into feat/accept-status-codes-backend

# Conflicts:
#	server/src/service/infrastructure/network/HttpProvider.ts

server/src/config/services.ts

@@ -107,6 +107,7 @@ import {
 	MongoTeamsRepository,
 	MongoMaintenanceWindowsRepository,
 	MongoSettingsRepository,
+	MongoQueueWorkersRepository,
 	IMonitorsRepository,
 	IChecksRepository,
 	IGeoChecksRepository,
@@ -123,7 +124,8 @@ import {
 	IMaintenanceWindowsRepository,
 } from "@/repositories/index.js";
 import { ILogger } from "@/utils/logger.js";
-// import { AppError } from "@/utils/AppError.js";
+import { AppError } from "@/utils/AppError.js";
+import { type QueueMode } from "@/types/settings.js";
 
 export type InitializedServices = {
 	settingsService: ISettingsService;
@@ -168,10 +170,12 @@ export const initializeServices = async ({
 	logger,
 	envSettings,
 	settingsService,
+	queueMode = "primary",
 }: {
 	logger: ILogger;
 	envSettings: EnvConfig;
 	settingsService: ISettingsService;
+	queueMode?: QueueMode;
 }): Promise<InitializedServices> => {
 	// Create DB
 
@@ -245,6 +249,7 @@ export const initializeServices = async ({
 	const incidentsRepository = new MongoIncidentsRepository();
 	const teamsRepository = new MongoTeamsRepository();
 	const maintenanceWindowsRepository = new MongoMaintenanceWindowsRepository();
+	const queueWorkersRepository = new MongoQueueWorkersRepository();
 
 	// Inject settings repository into settings service (now that DB is connected)
 	(settingsService as SettingsService).setRepository(settingsRepository);
@@ -345,10 +350,19 @@ export const initializeServices = async ({
 		geoChecksRepository
 	);
 
+	if (queueMode === "worker" && envSettings.queueType !== "lessSimpleQueue") {
+		throw new AppError({
+			message: `Worker mode requires QUEUE_TYPE="lessSimpleQueue" got ${envSettings.queueType}`,
+			status: 500,
+			service: "config services",
+			method: "initializeServices",
+		});
+	}
+
 	let jobQueue: IJobQueue;
 	switch (envSettings.queueType) {
 		case "lessSimpleQueue":
-			jobQueue = await LessSimpleQueue.create(logger, queueHelper, monitorsRepository, envSettings);
+			jobQueue = await LessSimpleQueue.create(logger, queueHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode);
 			break;
 		default:
 			jobQueue = await SuperSimpleQueue.create(logger, queueHelper, monitorsRepository);

server/src/controllers/queueController.ts

@@ -1,4 +1,5 @@
-import { ISuperSimpleQueue } from "@/service/index.js";
+import { IJobQueue } from "@/service/index.js";
+import { getQueueJobsQueryValidation } from "@/validation/queueValidation.js";
 import { Request, Response, NextFunction } from "express";
 
 const SERVICE_NAME = "JobQueueController";
@@ -12,8 +13,8 @@ export interface IJobQueueController {
 
 class JobQueueController implements IJobQueueController {
 	static SERVICE_NAME = SERVICE_NAME;
-	private jobQueue: ISuperSimpleQueue;
-	constructor(jobQueue: ISuperSimpleQueue) {
+	private jobQueue: IJobQueue;
+	constructor(jobQueue: IJobQueue) {
 		this.jobQueue = jobQueue;
 	}
 
@@ -36,11 +37,12 @@ class JobQueueController implements IJobQueueController {
 
 	getJobs = async (req: Request, res: Response, next: NextFunction) => {
 		try {
-			const jobs = await this.jobQueue.getJobs();
+			const pagination = getQueueJobsQueryValidation.parse(req.query);
+			const { jobs, count } = await this.jobQueue.getJobs(pagination);
 			return res.status(200).json({
 				success: true,
 				msg: "Queue jobs fetched successfully",
-				data: jobs,
+				data: { jobs, count },
 			});
 		} catch (error) {
 			next(error);
@@ -49,12 +51,13 @@ class JobQueueController implements IJobQueueController {
 
 	getAllMetrics = async (req: Request, res: Response, next: NextFunction) => {
 		try {
-			const jobs = await this.jobQueue.getJobs();
+			const pagination = getQueueJobsQueryValidation.parse(req.query);
+			const { jobs, count } = await this.jobQueue.getJobs(pagination);
 			const metrics = await this.jobQueue.getMetrics();
 			return res.status(200).json({
 				success: true,
 				msg: "Queue metrics fetched successfully",
-				data: { jobs, metrics },
+				data: { jobs, count, metrics },
 			});
 		} catch (error) {
 			next(error);

server/src/db/models/QueueWorker.ts

@@ -0,0 +1,28 @@
+import { Schema, model } from "mongoose";
+import { QueueModes, type QueueMode } from "@/types/settings.js";
+
+// Heartbeat presence record for a scheduler instance. _id is the scheduler's
+// workerId (hostname:pid:uuid). The TTL index on lastSeenAt garbage-collects
+// records for instances that have stopped heartbeating.
+interface QueueWorkerDocument {
+	_id: string;
+	mode: QueueMode;
+	lastSeenAt: Date;
+	createdAt: Date;
+	updatedAt: Date;
+}
+
+const QueueWorkerSchema = new Schema<QueueWorkerDocument>(
+	{
+		_id: { type: String, required: true },
+		mode: { type: String, enum: QueueModes, required: true },
+		lastSeenAt: { type: Date, default: Date.now, expires: 30 },
+	},
+	{ timestamps: true }
+);
+
+const QueueWorkerModel = model<QueueWorkerDocument>("QueueWorker", QueueWorkerSchema);
+
+export type { QueueWorkerDocument };
+export { QueueWorkerModel };
+export default QueueWorkerModel;

server/src/db/models/index.ts

@@ -41,3 +41,6 @@ export * from "@/db/models/DLQItem.js";
 export { default as DLQItemModel } from "@/db/models/DLQItem.js";
 export * from "@/db/models/Tag.js";
 export { default as TagModel } from "@/db/models/Tag.js";
+
+export * from "@/db/models/QueueWorker.js";
+export { default as QueueWorkerModel } from "@/db/models/QueueWorker.js";

server/src/index.ts

@@ -17,22 +17,31 @@ const startApp = async () => {
 	// Validate environment variables first
 	const env = validateEnv();
 
+	// Create settings service (env only — DB repository injected after connect)
+	const settingsService = new SettingsService(env);
+	const envSettings = settingsService.loadSettings();
+	// Create logger
+	logger = new Logger({ envSettings });
+
+	// Initialize services (connects DB, creates repositories, injects settingsRepository)
+	const services = await initializeServices({ logger, envSettings, settingsService, queueMode: envSettings.queueMode });
+
+	// If this is a worker instance, we're done.  No need for express
+	if (envSettings.queueMode === "worker") {
+		logger.info({
+			message: "Worker instance started. API will not be started",
+			service: SERVICE_NAME,
+		});
+		initShutdownListener(null, services);
+		return;
+	}
+
 	// FE path
 	const __filename = fileURLToPath(import.meta.url);
 	const __dirname = path.dirname(__filename);
 	const openApiSpec = JSON.parse(fs.readFileSync(path.join(__dirname, "../openapi.json"), "utf8"));
 	const frontendPath = path.join(__dirname, "..", "public");
 
-	// Create settings service (env only — DB repository injected after connect)
-	const settingsService = new SettingsService(env);
-	const envSettings = settingsService.loadSettings();
-
-	// Create logger
-	logger = new Logger({ envSettings });
-
-	// Initialize services (connects DB, creates repositories, injects settingsRepository)
-	const services = await initializeServices({ logger, envSettings, settingsService });
-
 	// Initialize controllers
 	const controllers = initializeControllers(services);
 

server/src/repositories/index.ts

@@ -42,3 +42,6 @@ export { default as MongoGeoChecksRepository } from "@/repositories/geo-checks/M
 
 export * from "@/repositories/dlq/IDLQRepository.js";
 export { default as MongoDLQRepository } from "@/repositories/dlq/MongoDLQRepository.js";
+
+export * from "@/repositories/queue-workers/IQueueWorkersRepository.js";
+export { default as MongoQueueWorkersRepository } from "@/repositories/queue-workers/MongoQueueWorkersRepository.js";

server/src/repositories/queue-workers/IQueueWorkersRepository.ts

@@ -0,0 +1,7 @@
+import type { QueueMode, QueueWorker } from "@/types/settings.js";
+
+export interface IQueueWorkersRepository {
+	upsert(workerId: string, mode: QueueMode): Promise<void>;
+	findRecent(maxAgeMs: number): Promise<QueueWorker[]>;
+	deleteById(workerId: string): Promise<void>;
+}

server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts

@@ -0,0 +1,33 @@
+import type { QueueMode, QueueWorker } from "@/types/settings.js";
+import type { IQueueWorkersRepository } from "./IQueueWorkersRepository.js";
+import type { QueueWorkerDocument } from "@/db/models/QueueWorker.js";
+import { QueueWorkerModel } from "@/db/models/QueueWorker.js";
+const SERVICE_NAME = "MongoQueueWorkersRepository";
+
+class MongoQueueWorkersRepository implements IQueueWorkersRepository {
+	static SERVICE_NAME = SERVICE_NAME;
+
+	protected toEntity = (doc: QueueWorkerDocument): QueueWorker => {
+		return {
+			workerId: doc._id,
+			mode: doc.mode,
+			lastSeenAt: doc.lastSeenAt.getTime(),
+		};
+	};
+
+	upsert = async (workerId: string, mode: QueueMode): Promise<void> => {
+		await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, lastSeenAt: new Date() } }, { upsert: true });
+	};
+
+	findRecent = async (maxAgeMs: number): Promise<QueueWorker[]> => {
+		const cutoff = new Date(Date.now() - maxAgeMs);
+		const docs = await QueueWorkerModel.find({ lastSeenAt: { $gte: cutoff } });
+		return docs.map(this.toEntity);
+	};
+
+	deleteById = async (workerId: string): Promise<void> => {
+		await QueueWorkerModel.deleteOne({ _id: workerId });
+	};
+}
+
+export default MongoQueueWorkersRepository;

server/src/service/infrastructure/JobQueues/IJobQueue.ts

@@ -1,4 +1,5 @@
 import { Monitor } from "@/types/monitor.js";
+import { QueueWorker } from "@/types/settings.js";
 
 export type QueueJobFailure = {
 	monitorId: string | number;
@@ -16,6 +17,7 @@ export type QueueMetrics = {
 	jobsWithFailures: QueueJobFailure[];
 	totalRuns: number;
 	totalFailures: number;
+	workers: QueueWorker[];
 };
 
 export type QueueJobSummary = {
@@ -26,6 +28,8 @@ export type QueueJobSummary = {
 	monitorGeoInterval: number | null;
 	monitorActive: boolean | null;
 	active: boolean;
+	lockedBy: string | null;
+	lockedUntil: number | null;
 	lockedAt: number | null;
 	runCount: number;
 	failCount: number;
@@ -37,6 +41,16 @@ export type QueueJobSummary = {
 	repeat: number | null;
 };
 
+export type QueueJobsPagination = {
+	page?: number;
+	rowsPerPage?: number;
+};
+
+export type QueueJobsPage = {
+	jobs: QueueJobSummary[];
+	count: number;
+};
+
 export interface IJobQueue {
 	readonly serviceName: string;
 	init(): Promise<boolean>;
@@ -47,6 +61,6 @@ export interface IJobQueue {
 	updateJob(monitor: Monitor): Promise<void>;
 	shutdown(): Promise<void>;
 	getMetrics(): Promise<QueueMetrics>;
-	getJobs(): Promise<QueueJobSummary[]>;
+	getJobs(pagination: QueueJobsPagination): Promise<QueueJobsPage>;
 	flushQueues(): Promise<{ success: boolean }>;
 }

server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts

@@ -1,12 +1,16 @@
-import { IMonitorsRepository } from "@/repositories/index.js";
+import { IMonitorsRepository, IQueueWorkersRepository } from "@/repositories/index.js";
 import { ILogger } from "@/utils/logger.js";
-import { MongoStore, Scheduler, IJob } from "less-simple-scheduler";
+import { MongoStore, Scheduler, IJob, AddJobInput } from "less-simple-scheduler";
 import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js";
 import { Monitor, supportsGeoCheck } from "@/types/monitor.js";
-import { type QueueMetrics, IJobQueue } from "@/service/infrastructure/JobQueues/IJobQueue.js";
+import { type QueueMetrics, type QueueJobsPagination, type QueueJobsPage, IJobQueue } from "@/service/infrastructure/JobQueues/IJobQueue.js";
+import { type QueueMode } from "@/types/settings.js";
 import { EnvConfig } from "@/service/system/settingsService.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"]);
@@ -18,13 +22,25 @@ export class LessSimpleQueue implements IJobQueue {
 	private logger: ILogger;
 	private helper: IQueueHelper;
 	private monitorsRepository: IMonitorsRepository;
+	private workersRepository: IQueueWorkersRepository;
 	private readonly scheduler: Scheduler;
+	private queueMode: QueueMode;
+	private heartbeatListener: ((workerId: string) => void) | null = null;
 
-	constructor(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) {
+	constructor(
+		logger: ILogger,
+		helper: IQueueHelper,
+		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();
 	}
 
@@ -49,7 +65,7 @@ export class LessSimpleQueue implements IJobQueue {
 
 		this.scheduler.on("scheduler:error", (workerId, error) => {
 			this.logger.error({
-				message: `w${workerId} error: ${error instanceof Error ? error.message : String(error)}`,
+				message: `${workerId} error: ${error instanceof Error ? error.message : String(error)}`,
 				service: SERVICE_NAME,
 				stack: error instanceof Error ? error.stack : undefined,
 			});
@@ -57,35 +73,35 @@ export class LessSimpleQueue implements IJobQueue {
 
 		this.scheduler.on("job:locked", (workerId, job) => {
 			this.logger.debug({
-				message: `w${workerId} ${job.id} locked`,
+				message: `${workerId} ${job.id} locked`,
 				service: SERVICE_NAME,
 			});
 		});
 
 		this.scheduler.on("job:abort", (workerId, job, reason) => {
 			this.logger.warn({
-				message: `w${workerId} ${job.id} aborted: ${reason}`,
+				message: `${workerId} ${job.id} aborted: ${reason}`,
 				service: SERVICE_NAME,
 			});
 		});
 
 		this.scheduler.on("job:attempt", (workerId, job, attempt) => {
 			this.logger.debug({
-				message: `w${workerId} ${job.id} attempt ${attempt}`,
+				message: `${workerId} ${job.id} attempt ${attempt}`,
 				service: SERVICE_NAME,
 			});
 		});
 
 		this.scheduler.on("job:complete", (workerId, job) => {
 			this.logger.debug({
-				message: `w${workerId} ${job.id} completed successfully`,
+				message: `${workerId} ${job.id} completed successfully`,
 				service: SERVICE_NAME,
 			});
 		});
 
 		this.scheduler.on("job:exhausted", (workerId, job, error) => {
 			this.logger.error({
-				message: `w${workerId} ${job.id} exhausted all retries: ${error instanceof Error ? error.message : String(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,
 			});
@@ -93,7 +109,7 @@ export class LessSimpleQueue implements IJobQueue {
 
 		this.scheduler.on("job:fail", (workerId, job, error, attempt) => {
 			this.logger.warn({
-				message: `w${workerId} ${job.id} failed on attempt ${attempt}: ${error instanceof Error ? error.message : String(error)}`,
+				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,
 			});
@@ -101,41 +117,80 @@ export class LessSimpleQueue implements IJobQueue {
 
 		this.scheduler.on("job:start", (workerId, job) => {
 			this.logger.debug({
-				message: `w${workerId} ${job.id} started`,
+				message: `${workerId} ${job.id} started`,
 				service: SERVICE_NAME,
 			});
 		});
 	};
 
-	static async create(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, envSettings: EnvConfig) {
+	static async create(
+		logger: ILogger,
+		helper: IQueueHelper,
+		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: 50,
+			concurrency: envSettings.queueMode === "primary" ? 0 : 50,
+			processEvery: 100,
 		});
-		const instance = new LessSimpleQueue(logger, helper, monitorsRepository, scheduler);
-		await instance.init();
+		const instance = new LessSimpleQueue(logger, helper, monitorsRepository, workersRepository, scheduler, queueMode);
+		await instance.init(queueMode);
 
 		return instance;
 	}
 
-	init = async () => {
+	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;
 			}
-			for (const monitor of monitors) {
-				const randomOffset = Math.floor(Math.random() * 100);
-				setTimeout(() => {
-					this.addJob(monitor.id, monitor);
-				}, randomOffset);
+			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 });
@@ -152,32 +207,38 @@ export class LessSimpleQueue implements IJobQueue {
 		}
 	};
 
-	addJob = async (monitorId: string, monitor: Monitor) => {
-		this.scheduler.addJob({
-			id: monitorId,
-			template: "monitor-job",
-			repeat: monitor.interval,
-			active: monitor.isActive,
-			data: monitor,
-			upsert: true,
-		});
+	// 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,
+			},
+		];
 
-		// 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`,
+		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) => {
@@ -272,56 +333,55 @@ export class LessSimpleQueue implements IJobQueue {
 	};
 
 	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 () => {
-		const jobs = await this.scheduler.getJobs();
-		const metrics = jobs.reduce<QueueMetrics>(
-			(acc, job) => {
-				const runCount = job.runCount ?? 0;
-				const failCount = job.failCount ?? 0;
-				const lastFailedAt = job.lastFailedAt ?? 0;
-				const lastFinishedAt = job.lastFinishedAt ?? 0;
-				acc.totalRuns += runCount;
-				acc.totalFailures += failCount;
-				acc.jobs++;
-				if (failCount > 0 && lastFailedAt >= lastFinishedAt) {
-					acc.failingJobs++;
-				}
+	getMetrics = async (): Promise<QueueMetrics> => {
+		const stats = await this.scheduler.getStats();
 
-				if (job.lockedAt) {
-					acc.activeJobs++;
-				}
+		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,
+			};
+		});
 
-				if (failCount > 0) {
-					const monitor = isMonitorJob(job) ? job.data : undefined;
-					acc.jobsWithFailures.push({
-						monitorId: job.id,
-						monitorUrl: monitor?.url || null,
-						monitorType: monitor?.type || null,
-						failedAt: job.lastFailedAt ?? null,
-						failCount,
-						failReason: job.lastError ?? null,
-					});
-				}
-				return acc;
-			},
-			{
-				jobs: 0,
-				activeJobs: 0,
-				failingJobs: 0,
-				jobsWithFailures: [],
-				totalRuns: 0,
-				totalFailures: 0,
-			}
-		);
-		return metrics;
+		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 () => {
-		const jobs = await this.scheduler.getJobs();
-		return jobs.map((job) => {
+	getJobs = async ({ page = 0, rowsPerPage = 0 }: QueueJobsPagination): Promise<QueueJobsPage> => {
+		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 {
@@ -333,6 +393,8 @@ export class LessSimpleQueue implements IJobQueue {
 				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,
@@ -343,11 +405,19 @@ export class LessSimpleQueue implements IJobQueue {
 				lastFailedAt: job.lastFailedAt ?? null,
 			};
 		});
+
+		return { jobs: mapped, count };
 	};
 
 	flushQueues = async () => {
-		const stopRes = await this.scheduler.stop();
-		const flushRes = await this.scheduler.flushJobs();
+		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/service/infrastructure/JobQueues/SuperSimpleQueue.ts

@@ -3,64 +3,19 @@ import { ILogger } from "@/utils/logger.js";
 import Scheduler from "super-simple-scheduler";
 import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js";
 import { Monitor, MonitorType, supportsGeoCheck } from "@/types/monitor.js";
+import { hostname } from "node:os";
+import { randomUUID } from "node:crypto";
+import { QueueMode } from "@/types/settings.js";
+import { IJobQueue, type QueueJobSummary } from "@/service/infrastructure/JobQueues/IJobQueue.js";
+
 const SERVICE_NAME = "JobQueue";
 
-type QueueJobFailure = {
-	monitorId: string | number;
-	monitorUrl: string | null;
-	monitorType: string | null;
-	failedAt: number | null;
-	failCount: number;
-	failReason: string | null;
-};
-
-type QueueMetrics = {
-	jobs: number;
-	activeJobs: number;
-	failingJobs: number;
-	jobsWithFailures: QueueJobFailure[];
-	totalRuns: number;
-	totalFailures: number;
-};
-
-type QueueJobSummary = {
-	monitorId: string | number;
-	monitorUrl: string | null;
-	monitorType: string | null;
-	monitorInterval: number | null;
-	monitorGeoInterval: number | null;
-	monitorActive: boolean | null;
-	active: boolean;
-	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 interface ISuperSimpleQueue {
-	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<QueueMetrics>;
-	getJobs(): Promise<QueueJobSummary[]>;
-	flushQueues(): Promise<{ success: boolean }>;
-}
-
-export class SuperSimpleQueue implements ISuperSimpleQueue {
+export class SuperSimpleQueue implements IJobQueue {
 	static SERVICE_NAME = SERVICE_NAME;
 
 	private logger: ILogger;
 	private helper: IQueueHelper;
+	private workerId: string;
 	private monitorsRepository: IMonitorsRepository;
 	private readonly scheduler: Scheduler;
 
@@ -70,6 +25,7 @@ export class SuperSimpleQueue implements ISuperSimpleQueue {
 		this.monitorsRepository = monitorsRepository;
 		this.scheduler = scheduler;
 		this.registerListeners();
+		this.workerId = `${hostname()}:${process.pid}:${randomUUID()}`;
 	}
 
 	get serviceName() {
@@ -311,7 +267,26 @@ export class SuperSimpleQueue implements ISuperSimpleQueue {
 
 	getMetrics = async () => {
 		const jobs = await this.scheduler.getJobs();
-		const metrics = jobs.reduce(
+		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;
@@ -354,14 +329,21 @@ export class SuperSimpleQueue implements ISuperSimpleQueue {
 				}>,
 				totalRuns: 0,
 				totalFailures: 0,
+				workers: [
+					{
+						workerId: this.workerId,
+						mode: "primary",
+						lastSeenAt: Date.now(),
+					},
+				],
 			}
 		);
 		return metrics;
 	};
 
-	getJobs = async () => {
+	getJobs = async ({ page = 0, rowsPerPage = 0 }: { page?: number; rowsPerPage?: number }): Promise<{ jobs: QueueJobSummary[]; count: number }> => {
 		const jobs = await this.scheduler.getJobs();
-		return jobs.map((job) => {
+		const mapped = jobs.map((job) => {
 			return {
 				monitorId: job.id,
 				monitorUrl: job.data?.url || null,
@@ -371,6 +353,8 @@ export class SuperSimpleQueue implements ISuperSimpleQueue {
 				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,
@@ -381,6 +365,12 @@ export class SuperSimpleQueue implements ISuperSimpleQueue {
 				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 () => {

server/src/service/infrastructure/bufferService.ts

@@ -26,7 +26,7 @@ export class BufferService implements IBufferService {
 	private geoChecksService: IGeoChecksService;
 
 	constructor(logger: ILogger, checkService: ICheckService, geoChecksService: IGeoChecksService, settingsService: ISettingsService) {
-		this.BUFFER_TIMEOUT = settingsService.getSettings().nodeEnv === "development" ? 10 : 1000 * 60 * 1; // 1 minute
+		this.BUFFER_TIMEOUT = settingsService.getSettings().nodeEnv === "development" ? 1000 : 1000 * 60 * 1; // 1 minute
 		this.logger = logger;
 		this.checksService = checkService;
 		this.geoChecksService = geoChecksService;

server/src/service/infrastructure/network/HttpProvider.ts

@@ -4,6 +4,7 @@ import { IStatusProvider } from "@/service/infrastructure/network/IStatusProvide
 import { HttpStatusPayload } from "@/types/network.js";
 import { MonitorStatusResponse } from "@/types/network.js";
 import { Agent as HttpsAgent } from "https";
+import { Agent as HttpAgent } from "http";
 import { Monitor, MonitorType } from "@/types/monitor.js";
 import { NETWORK_ERROR, isStatusUp } from "@/service/infrastructure/network/utils.js";
 import CacheableLookup from "cacheable-lookup";
@@ -11,6 +12,11 @@ import CacheableLookup from "cacheable-lookup";
 export class HttpProvider implements IStatusProvider<HttpStatusPayload> {
 	readonly type = "http";
 
+	// Shared, pooled agents reused across every check
+	private readonly httpAgent: HttpAgent;
+	private readonly httpsAgent: HttpsAgent;
+	private readonly httpsAgentInsecure: HttpsAgent;
+
 	constructor(
 		private got: Got,
 		private advancedMatcher: IAdvancedMatcher
@@ -23,6 +29,11 @@ export class HttpProvider implements IStatusProvider<HttpStatusPayload> {
 			},
 			retry: { limit: 1 },
 		});
+
+		const agentOptions = { keepAlive: true, maxSockets: 256, maxFreeSockets: 256 };
+		this.httpAgent = new HttpAgent(agentOptions);
+		this.httpsAgent = new HttpsAgent({ ...agentOptions, rejectUnauthorized: true });
+		this.httpsAgentInsecure = new HttpsAgent({ ...agentOptions, rejectUnauthorized: false });
 	}
 
 	supports(type: MonitorType) {
@@ -33,7 +44,7 @@ export class HttpProvider implements IStatusProvider<HttpStatusPayload> {
 		if (error instanceof HTTPError || error instanceof RequestError) {
 			const statusCode = error.response?.statusCode;
 			const statusUp = isStatusUp(statusCode, monitor.customUpCodes);
-			const responseTime = error.timings?.phases?.total ?? 0;
+			const responseTime = error.timings?.phases?.firstByte ?? error.timings?.phases?.total ?? 0;
 
 			if (!statusUp) {
 				return {
@@ -119,7 +130,8 @@ export class HttpProvider implements IStatusProvider<HttpStatusPayload> {
 		};
 
 		options.agent = {
-			https: new HttpsAgent({ rejectUnauthorized: !ignoreTlsErrors }),
+			http: this.httpAgent,
+			https: ignoreTlsErrors ? this.httpsAgentInsecure : this.httpsAgent,
 		};
 
 		try {
@@ -135,7 +147,7 @@ export class HttpProvider implements IStatusProvider<HttpStatusPayload> {
 					status: false,
 					code: response.statusCode,
 					message: "Response is not JSON",
-					responseTime: response.timings.phases.total ?? 0,
+					responseTime: response.timings.phases.firstByte ?? 0,
 					timings: response.timings,
 					payload: response.body as unknown as T,
 				};
@@ -161,7 +173,7 @@ export class HttpProvider implements IStatusProvider<HttpStatusPayload> {
 				status: statusUp && matchResult.ok,
 				code: response.statusCode,
 				message: matchResult.ok ? (response.statusMessage ?? "OK") : matchResult.message,
-				responseTime: response.timings.phases.total ?? 0,
+				responseTime: response.timings.phases.firstByte ?? 0,
 				timings: response.timings,
 				payload,
 				extracted: matchResult.extracted,

server/src/service/system/settingsService.ts

@@ -1,5 +1,5 @@
 import { ISettingsRepository } from "@/repositories/index.js";
-import { type DbType, type QueueType, Settings, SettingsUpdate } from "@/types/index.js";
+import { type DbType, type QueueType, type QueueMode, Settings, SettingsUpdate } from "@/types/index.js";
 import { AppError } from "@/utils/AppError.js";
 import { ValidatedEnv } from "@/validation/envValidation.js";
 import type { StringValue } from "ms";
@@ -14,6 +14,7 @@ export type EnvConfig = {
 	dbConnectionString: string;
 	dbType: DbType;
 	queueType: QueueType;
+	queueMode: QueueMode;
 	statusPageThemesEnabled: boolean;
 };
 
@@ -41,6 +42,7 @@ export class SettingsService implements ISettingsService {
 			dbConnectionString: env.DB_CONNECTION_STRING,
 			dbType: env.DB_TYPE,
 			queueType: env.QUEUE_TYPE,
+			queueMode: env.QUEUE_MODE,
 			statusPageThemesEnabled: env.STATUS_PAGE_THEMES_ENABLED,
 		};
 	}

server/src/shutdown.ts

@@ -2,7 +2,7 @@ import { InitializedServices } from "./config/services.js";
 import { logger } from "./utils/logger.js";
 import type { Server } from "http";
 
-export const initShutdownListener = (server: Server, services: InitializedServices) => {
+export const initShutdownListener = (server: Server | null, services: InitializedServices) => {
 	const SERVICE_NAME = "Server";
 
 	let isShuttingDown = false;
@@ -15,7 +15,7 @@ export const initShutdownListener = (server: Server, services: InitializedServic
 		logger.info({ message: "Attempting graceful shutdown" });
 
 		try {
-			server.close();
+			server?.close();
 			await services.jobQueue.shutdown();
 			await services.db.disconnect();
 			logger.info({ message: "Graceful shutdown complete" });

server/src/types/settings.ts

@@ -4,6 +4,15 @@ export type DbType = (typeof DbTypes)[number];
 export const QueueTypes = ["lessSimpleQueue", "superSimpleQueue"] as const;
 export type QueueType = (typeof QueueTypes)[number];
 
+export const QueueModes = ["primary", "worker"] as const;
+export type QueueMode = (typeof QueueModes)[number];
+
+export type QueueWorker = {
+	workerId: string; // hostname:pid:uuid
+	mode: QueueMode;
+	lastSeenAt: number; // epoch ms of the last heartbeat
+};
+
 export interface SettingsThresholds {
 	cpu?: number;
 	memory?: number;

server/src/validation/envValidation.ts

@@ -1,5 +1,5 @@
 import { z } from "zod";
-import { DbTypes, QueueTypes } from "@/types/settings.js";
+import { DbTypes, QueueTypes, QueueModes } from "@/types/settings.js";
 import { booleanCoercion } from "./shared.js";
 
 const envSchema = z.object({
@@ -11,7 +11,9 @@ const envSchema = z.object({
 	// Database
 	DB_CONNECTION_STRING: z.string().min(1, "Database connection string is required"),
 	DB_TYPE: z.enum(DbTypes).default("mongodb"),
+
 	QUEUE_TYPE: z.enum(QueueTypes).default("superSimpleQueue"),
+	QUEUE_MODE: z.enum(QueueModes).default("primary"),
 
 	// JWT Authentication
 	JWT_SECRET: z.string().min(1, "JWT_SECRET is required"),

server/src/validation/queueValidation.ts

@@ -0,0 +1,6 @@
+import { z } from "zod";
+
+export const getQueueJobsQueryValidation = z.object({
+	page: z.coerce.number().int().min(0).optional(),
+	rowsPerPage: z.coerce.number().int().min(1).max(100).optional(),
+});