Merge pull request #3691 from bluewave-labs/feat/workers

feat: workers

server/package.json

@@ -9,8 +9,10 @@
 		"test:unit": "NODE_OPTIONS=--experimental-vm-modules jest --coverage --collectCoverageFrom='src/**/*.ts' test/unit",
 		"test:unit:services": "NODE_OPTIONS=--experimental-vm-modules jest --coverage --collectCoverageFrom='src/service/**/*.ts' test/unit/services test/unit/providers/network test/unit/providers/notifications",
 		"test:integration": "NODE_OPTIONS=--experimental-vm-modules jest --no-coverage test/integration",
-		"dev": "nodemon --exec tsx src/index.js",
 		"start": "node --watch ./dist/index.js",
+		"dev": "nodemon --exec tsx src/index.js",
+		"worker": "QUEUE_MODE=worker node ./dist/index.js",
+		"worker:dev": "QUEUE_MODE=worker nodemon --exec tsx src/index.js",
 		"clean": "rm -rf dist",
 		"build": "npm run clean && npm run build:openapi && tsc -p tsconfig.build.json && tsc-alias -p tsconfig.build.json && cp -r src/templates dist/templates",
 		"build:openapi": "tsc --noEmit && tsx scripts/buildOpenApi.ts",

server/src/config/services.ts

@@ -123,7 +123,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 +169,12 @@ export const initializeServices = async ({
 	logger,
 	envSettings,
 	settingsService,
+	queueMode = "primary",
 }: {
 	logger: ILogger;
 	envSettings: EnvConfig;
 	settingsService: ISettingsService;
+	queueMode?: QueueMode;
 }): Promise<InitializedServices> => {
 	// Create DB
 
@@ -345,10 +348,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, envSettings, queueMode);
 			break;
 		default:
 			jobQueue = await SuperSimpleQueue.create(logger, queueHelper, monitorsRepository);

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/service/infrastructure/JobQueues/LessSimpleQueue.ts

@@ -4,6 +4,7 @@ import { MongoStore, Scheduler, IJob } 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 QueueMode } from "@/types/settings.js";
 import { EnvConfig } from "@/service/system/settingsService.js";
 const SERVICE_NAME = "JobQueue";
 
@@ -49,7 +50,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 +58,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 +94,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,13 +102,13 @@ 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, envSettings: EnvConfig, queueMode: QueueMode) {
 		const store = new MongoStore({
 			url: envSettings.dbConnectionString ?? "mongodb://localhost:27017/uptime_db",
 		});
@@ -115,18 +116,29 @@ export class LessSimpleQueue implements IJobQueue {
 			concurrency: 50,
 		});
 		const instance = new LessSimpleQueue(logger, helper, monitorsRepository, scheduler);
-		await instance.init();
+		await instance.init(queueMode);
 
 		return instance;
 	}
 
-	init = async () => {
+	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());
+
+			// 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;

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,9 @@ 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 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/test/unit/services/lessSimpleQueue.test.ts

@@ -0,0 +1,652 @@
+import { describe, expect, it, jest } from "@jest/globals";
+import { createMockLogger } from "../../helpers/createMockLogger.ts";
+import type { Monitor } from "../../../src/types/monitor.ts";
+
+// ── Mock Scheduler / MongoStore ──────────────────────────────────────────────
+
+const createScheduler = () => {
+	const listeners: Record<string, Function[]> = {};
+	return {
+		workerId: "host-1:1234:abcd",
+		on: jest.fn((event: string, cb: Function) => {
+			listeners[event] = listeners[event] || [];
+			listeners[event].push(cb);
+		}),
+		emit: (event: string, ...args: unknown[]) => {
+			(listeners[event] || []).forEach((cb) => cb(...args));
+		},
+		start: jest.fn().mockResolvedValue(true),
+		stop: jest.fn().mockResolvedValue(true),
+		addTemplate: jest.fn(),
+		addJob: jest.fn(),
+		removeJob: jest.fn(),
+		getJob: jest.fn().mockResolvedValue(null),
+		getJobs: jest.fn().mockResolvedValue([]),
+		pauseJob: jest.fn().mockResolvedValue(true),
+		resumeJob: jest.fn().mockResolvedValue(true),
+		updateJob: jest.fn(),
+		flushJobs: jest.fn().mockResolvedValue(true),
+	};
+};
+
+const mockSchedulerInstance = createScheduler();
+const MockScheduler = jest.fn().mockReturnValue(mockSchedulerInstance);
+const MockMongoStore = jest.fn();
+
+jest.unstable_mockModule("less-simple-scheduler", () => ({
+	Scheduler: MockScheduler,
+	MongoStore: MockMongoStore,
+}));
+
+const { LessSimpleQueue } = await import("../../../src/service/infrastructure/JobQueues/LessSimpleQueue.ts");
+
+const createQueueHelper = () => ({
+	getHeartbeatJob: jest.fn().mockReturnValue(() => Promise.resolve()),
+	getHeartbeatGeoJob: jest.fn().mockReturnValue(() => Promise.resolve()),
+	getCleanupOrphanedJob: jest.fn().mockReturnValue(() => Promise.resolve()),
+	getCleanupRetentionJob: jest.fn().mockReturnValue(() => Promise.resolve()),
+});
+
+const createMonitorsRepo = () => ({
+	findAll: jest.fn().mockResolvedValue([]),
+});
+
+const createQueue = (overrides?: Record<string, unknown>) => {
+	const logger = createMockLogger();
+	const helper = createQueueHelper();
+	const monitorsRepository = createMonitorsRepo();
+	const scheduler = createScheduler();
+
+	const defaults = { logger, helper, monitorsRepository, scheduler, ...overrides };
+
+	const queue = new LessSimpleQueue(defaults.logger as any, defaults.helper as any, defaults.monitorsRepository as any, defaults.scheduler as any);
+	return { queue, ...defaults };
+};
+
+const makeMonitor = (overrides?: Partial<Monitor>): Monitor =>
+	({
+		id: "mon-1",
+		teamId: "team-1",
+		type: "http",
+		interval: 60000,
+		isActive: true,
+		geoCheckEnabled: false,
+		geoCheckInterval: 300000,
+		geoCheckLocations: [],
+		...overrides,
+	}) as Monitor;
+
+// ── Tests ────────────────────────────────────────────────────────────────────
+
+describe("LessSimpleQueue", () => {
+	describe("serviceName", () => {
+		it("returns JobQueue", () => {
+			const { queue } = createQueue();
+			expect(queue.serviceName).toBe("JobQueue");
+		});
+	});
+
+	// ── init (primary) ─────────────────────────────────────────────────────────
+
+	describe("init - primary mode", () => {
+		it("starts scheduler, registers all templates, seeds monitors and system jobs", async () => {
+			const { queue, scheduler, monitorsRepository } = createQueue();
+
+			const result = await queue.init();
+
+			expect(result).toBe(true);
+			expect(scheduler.start).toHaveBeenCalled();
+			expect(scheduler.addTemplate).toHaveBeenCalledWith("monitor-job", expect.any(Function));
+			expect(scheduler.addTemplate).toHaveBeenCalledWith("geo-check-job", expect.any(Function));
+			expect(scheduler.addTemplate).toHaveBeenCalledWith("cleanup-orphaned", expect.any(Function));
+			expect(scheduler.addTemplate).toHaveBeenCalledWith("cleanup-retention-job", expect.any(Function));
+			expect(monitorsRepository.findAll).toHaveBeenCalled();
+			expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-orphaned" }));
+			expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-retention" }));
+		});
+
+		it("schedules existing monitors with staggered offsets", async () => {
+			jest.useFakeTimers();
+			const monitors = [makeMonitor({ id: "m1" }), makeMonitor({ id: "m2" })];
+			const { queue, scheduler, monitorsRepository } = createQueue();
+			(monitorsRepository.findAll as jest.Mock).mockResolvedValue(monitors);
+
+			await queue.init();
+			jest.runAllTimers();
+
+			expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "m1", template: "monitor-job" }));
+			expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "m2", template: "monitor-job" }));
+			jest.useRealTimers();
+		});
+
+		it("returns true and skips system jobs when findAll returns null", async () => {
+			const { queue, monitorsRepository, scheduler } = createQueue();
+			(monitorsRepository.findAll as jest.Mock).mockResolvedValue(null);
+
+			const result = await queue.init();
+
+			expect(result).toBe(true);
+			expect(scheduler.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-orphaned" }));
+		});
+
+		it("returns false and logs error on failure", async () => {
+			const { queue, logger, monitorsRepository } = createQueue();
+			(monitorsRepository.findAll as jest.Mock).mockRejectedValue(new Error("db down"));
+
+			const result = await queue.init();
+
+			expect(result).toBe(false);
+			expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("db down") }));
+		});
+
+		it("logs error with String(error) for non-Error exceptions", async () => {
+			const { queue, logger, monitorsRepository } = createQueue();
+			(monitorsRepository.findAll as jest.Mock).mockRejectedValue("string error");
+
+			const result = await queue.init();
+
+			expect(result).toBe(false);
+			expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("string error") }));
+		});
+	});
+
+	// ── init (worker) ──────────────────────────────────────────────────────────
+
+	describe("init - worker mode", () => {
+		it("starts scheduler and registers all templates", async () => {
+			const { queue, scheduler } = createQueue();
+
+			const result = await queue.init("worker");
+
+			expect(result).toBe(true);
+			expect(scheduler.start).toHaveBeenCalled();
+			expect(scheduler.addTemplate).toHaveBeenCalledWith("monitor-job", expect.any(Function));
+			expect(scheduler.addTemplate).toHaveBeenCalledWith("geo-check-job", expect.any(Function));
+			expect(scheduler.addTemplate).toHaveBeenCalledWith("cleanup-orphaned", expect.any(Function));
+			expect(scheduler.addTemplate).toHaveBeenCalledWith("cleanup-retention-job", expect.any(Function));
+		});
+
+		it("does NOT seed monitor jobs (no findAll, primary populates the queue)", async () => {
+			const { queue, monitorsRepository } = createQueue();
+
+			await queue.init("worker");
+
+			expect(monitorsRepository.findAll).not.toHaveBeenCalled();
+		});
+
+		it("does NOT seed cleanup system jobs", async () => {
+			const { queue, scheduler } = createQueue();
+
+			await queue.init("worker");
+
+			expect(scheduler.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-orphaned" }));
+			expect(scheduler.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "cleanup-retention" }));
+			expect(scheduler.addJob).not.toHaveBeenCalled();
+		});
+
+		it("logs the initialized worker with its workerId", async () => {
+			const { queue, logger } = createQueue();
+
+			await queue.init("worker");
+
+			expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: "Worker queue: host-1:1234:abcd initialized" }));
+		});
+	});
+
+	// ── registerListeners ──────────────────────────────────────────────────────
+
+	describe("registerListeners (via constructor)", () => {
+		it("logs on scheduler:start", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("scheduler:start", "host-1:1234:abcd");
+			expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: "Scheduler started" }));
+		});
+
+		it("logs on scheduler:stop", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("scheduler:stop", "host-1:1234:abcd");
+			expect(logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: "Scheduler stopped" }));
+		});
+
+		it("logs on scheduler:error with Error", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("scheduler:error", "w1", new Error("boom"));
+			expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 error: boom" }));
+		});
+
+		it("logs on scheduler:error with non-Error", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("scheduler:error", "w1", "string error");
+			expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 error: string error", stack: undefined }));
+		});
+
+		it("logs on job:locked", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("job:locked", "w1", { id: "m1" });
+			expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 locked" }));
+		});
+
+		it("logs on job:abort", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("job:abort", "w1", { id: "m1" }, "timeout");
+			expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 aborted: timeout" }));
+		});
+
+		it("logs on job:attempt", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("job:attempt", "w1", { id: "m1" }, 2);
+			expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 attempt 2" }));
+		});
+
+		it("logs on job:complete", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("job:complete", "w1", { id: "m1" });
+			expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 completed successfully" }));
+		});
+
+		it("logs on job:exhausted with Error", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("job:exhausted", "w1", { id: "m1" }, new Error("gave up"));
+			expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 exhausted all retries: gave up" }));
+		});
+
+		it("logs on job:exhausted with non-Error", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("job:exhausted", "w1", { id: "m1" }, "gave up");
+			expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 exhausted all retries: gave up", stack: undefined }));
+		});
+
+		it("logs on job:fail with Error", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("job:fail", "w1", { id: "m1" }, new Error("oops"), 1);
+			expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 failed on attempt 1: oops" }));
+		});
+
+		it("logs on job:fail with non-Error", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("job:fail", "w1", { id: "m1" }, "oops", 1);
+			expect(logger.warn).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 failed on attempt 1: oops", stack: undefined }));
+		});
+
+		it("logs on job:start", () => {
+			const { scheduler, logger } = createQueue();
+			scheduler.emit("job:start", "w1", { id: "m1" });
+			expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "w1 m1 started" }));
+		});
+	});
+
+	// ── addJob ─────────────────────────────────────────────────────────────────
+
+	describe("addJob", () => {
+		it("adds monitor job to scheduler", async () => {
+			const { queue, scheduler } = createQueue();
+			await queue.addJob("mon-1", makeMonitor());
+			expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1", template: "monitor-job", repeat: 60000, active: true }));
+		});
+
+		it("adds geo check job when geoCheckEnabled and type supports it", async () => {
+			const { queue, scheduler } = createQueue();
+			await queue.addJob("mon-1", makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckInterval: 300000 }));
+			expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1-geo", template: "geo-check-job", repeat: 300000 }));
+		});
+
+		it("skips geo job when type does not support geo checks", async () => {
+			const { queue, scheduler } = createQueue();
+			await queue.addJob("mon-1", makeMonitor({ geoCheckEnabled: true, type: "hardware" }));
+			expect(scheduler.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1-geo" }));
+		});
+
+		it("skips geo job when geoCheckEnabled is false", async () => {
+			const { queue, scheduler } = createQueue();
+			await queue.addJob("mon-1", makeMonitor({ geoCheckEnabled: false, type: "http" }));
+			expect(scheduler.addJob).not.toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1-geo" }));
+		});
+	});
+
+	// ── deleteJob ──────────────────────────────────────────────────────────────
+
+	describe("deleteJob", () => {
+		it("removes monitor job", async () => {
+			const { queue, scheduler } = createQueue();
+			await queue.deleteJob(makeMonitor());
+			expect(scheduler.removeJob).toHaveBeenCalledWith("mon-1");
+		});
+
+		it("removes geo job if it exists", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" });
+			await queue.deleteJob(makeMonitor());
+			expect(scheduler.removeJob).toHaveBeenCalledWith("mon-1-geo");
+		});
+
+		it("does not remove geo job if it does not exist", async () => {
+			const { queue, scheduler } = createQueue();
+			await queue.deleteJob(makeMonitor());
+			expect(scheduler.removeJob).not.toHaveBeenCalledWith("mon-1-geo");
+		});
+	});
+
+	// ── pauseJob ───────────────────────────────────────────────────────────────
+
+	describe("pauseJob", () => {
+		it("pauses monitor and geo job when geo exists", async () => {
+			const { queue, scheduler, logger } = createQueue();
+			(scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" });
+			await queue.pauseJob(makeMonitor());
+
+			expect(scheduler.pauseJob).toHaveBeenCalledWith("mon-1");
+			expect(scheduler.pauseJob).toHaveBeenCalledWith("mon-1-geo");
+			expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "Paused monitor mon-1" }));
+		});
+
+		it("does not pause geo job when it does not exist", async () => {
+			const { queue, scheduler } = createQueue();
+			await queue.pauseJob(makeMonitor());
+			expect(scheduler.pauseJob).toHaveBeenCalledWith("mon-1");
+			expect(scheduler.pauseJob).not.toHaveBeenCalledWith("mon-1-geo");
+		});
+
+		it("throws when scheduler.pauseJob returns false for the main job", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.pauseJob as jest.Mock).mockResolvedValue(false);
+			await expect(queue.pauseJob(makeMonitor())).rejects.toThrow("Failed to pause monitor");
+		});
+
+		it("logs error but does not throw when geo pauseJob returns false", async () => {
+			const { queue, scheduler, logger } = createQueue();
+			(scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" });
+			(scheduler.pauseJob as jest.Mock).mockImplementation(async (id: string) => id !== "mon-1-geo");
+
+			await expect(queue.pauseJob(makeMonitor())).resolves.toBeUndefined();
+			expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "Failed to pause geo check job for monitor mon-1" }));
+		});
+	});
+
+	// ── resumeJob ──────────────────────────────────────────────────────────────
+
+	describe("resumeJob", () => {
+		it("resumes monitor and geo job when geo exists", async () => {
+			const { queue, scheduler, logger } = createQueue();
+			(scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" });
+			await queue.resumeJob(makeMonitor());
+
+			expect(scheduler.resumeJob).toHaveBeenCalledWith("mon-1");
+			expect(scheduler.resumeJob).toHaveBeenCalledWith("mon-1-geo");
+			expect(logger.debug).toHaveBeenCalledWith(expect.objectContaining({ message: "Resumed monitor mon-1" }));
+		});
+
+		it("does not resume geo job when it does not exist", async () => {
+			const { queue, scheduler } = createQueue();
+			await queue.resumeJob(makeMonitor());
+			expect(scheduler.resumeJob).toHaveBeenCalledWith("mon-1");
+			expect(scheduler.resumeJob).not.toHaveBeenCalledWith("mon-1-geo");
+		});
+
+		it("throws when scheduler.resumeJob returns false for the main job", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.resumeJob as jest.Mock).mockResolvedValue(false);
+			await expect(queue.resumeJob(makeMonitor())).rejects.toThrow("Failed to resume monitor");
+		});
+
+		it("logs error but does not throw when geo resumeJob returns false", async () => {
+			const { queue, scheduler, logger } = createQueue();
+			(scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" });
+			(scheduler.resumeJob as jest.Mock).mockImplementation(async (id: string) => id !== "mon-1-geo");
+
+			await expect(queue.resumeJob(makeMonitor())).resolves.toBeUndefined();
+			expect(logger.error).toHaveBeenCalledWith(expect.objectContaining({ message: "Failed to resume geo check job for monitor mon-1" }));
+		});
+	});
+
+	// ── updateJob / syncGeoJob ─────────────────────────────────────────────────
+
+	describe("updateJob", () => {
+		it("updates monitor job with new interval", async () => {
+			const { queue, scheduler } = createQueue();
+			const monitor = makeMonitor({ interval: 120000 });
+			await queue.updateJob(monitor);
+			expect(scheduler.updateJob).toHaveBeenCalledWith("mon-1", expect.objectContaining({ repeat: 120000, data: monitor }));
+		});
+
+		it("removes existing geo job when geoCheckEnabled is false", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" });
+			await queue.updateJob(makeMonitor({ geoCheckEnabled: false, type: "http" }));
+			expect(scheduler.removeJob).toHaveBeenCalledWith("mon-1-geo");
+		});
+
+		it("removes existing geo job when type does not support geo checks", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" });
+			await queue.updateJob(makeMonitor({ geoCheckEnabled: true, type: "hardware" }));
+			expect(scheduler.removeJob).toHaveBeenCalledWith("mon-1-geo");
+		});
+
+		it("updates existing geo job when geoCheckEnabled", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJob as jest.Mock).mockResolvedValue({ id: "mon-1-geo" });
+			const monitor = makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckInterval: 300000 });
+			await queue.updateJob(monitor);
+			expect(scheduler.updateJob).toHaveBeenCalledWith("mon-1-geo", expect.objectContaining({ repeat: 300000 }));
+		});
+
+		it("creates new geo job when enabled but no existing job", async () => {
+			const { queue, scheduler } = createQueue();
+			const monitor = makeMonitor({ geoCheckEnabled: true, type: "http", geoCheckInterval: 300000 });
+			await queue.updateJob(monitor);
+			expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "mon-1-geo", template: "geo-check-job" }));
+		});
+	});
+
+	// ── shutdown ───────────────────────────────────────────────────────────────
+
+	describe("shutdown", () => {
+		it("stops the scheduler", async () => {
+			const { queue, scheduler } = createQueue();
+			await queue.shutdown();
+			expect(scheduler.stop).toHaveBeenCalled();
+		});
+	});
+
+	// ── getMetrics ─────────────────────────────────────────────────────────────
+
+	describe("getMetrics", () => {
+		it("returns empty metrics when no jobs", async () => {
+			const { queue } = createQueue();
+			const metrics = await queue.getMetrics();
+			expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0 });
+		});
+
+		it("aggregates metrics from jobs", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJobs as jest.Mock).mockResolvedValue([
+				{
+					id: "m1",
+					template: "monitor-job",
+					runCount: 10,
+					failCount: 2,
+					lastFailedAt: 200,
+					lastFinishedAt: 100,
+					lockedAt: null,
+					lastError: "timeout",
+					data: { url: "http://a.com", type: "http" },
+				},
+				{
+					id: "m2",
+					template: "monitor-job",
+					runCount: 5,
+					failCount: 0,
+					lastFailedAt: 0,
+					lastFinishedAt: 50,
+					lockedAt: 999,
+					data: { url: "http://b.com", type: "ping" },
+				},
+			]);
+			const metrics = await queue.getMetrics();
+
+			expect(metrics.jobs).toBe(2);
+			expect(metrics.totalRuns).toBe(15);
+			expect(metrics.totalFailures).toBe(2);
+			expect(metrics.activeJobs).toBe(1);
+			expect(metrics.failingJobs).toBe(1);
+			expect(metrics.jobsWithFailures).toHaveLength(1);
+			expect(metrics.jobsWithFailures[0]).toEqual(
+				expect.objectContaining({ monitorId: "m1", failCount: 2, monitorUrl: "http://a.com", failReason: "timeout" })
+			);
+		});
+
+		it("does not count as failing when lastFailedAt < lastFinishedAt", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJobs as jest.Mock).mockResolvedValue([
+				{ id: "m1", template: "monitor-job", runCount: 10, failCount: 1, lastFailedAt: 50, lastFinishedAt: 100, data: { url: "a" } },
+			]);
+			const metrics = await queue.getMetrics();
+			expect(metrics.failingJobs).toBe(0);
+			expect(metrics.jobsWithFailures).toHaveLength(1);
+		});
+
+		it("leaves monitor fields null for non-monitor (cleanup) jobs with failures", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJobs as jest.Mock).mockResolvedValue([
+				{
+					id: "cleanup-orphaned",
+					template: "cleanup-orphaned",
+					runCount: 1,
+					failCount: 1,
+					lastFailedAt: 100,
+					lastFinishedAt: 50,
+					lastError: "boom",
+					data: undefined,
+				},
+			]);
+			const metrics = await queue.getMetrics();
+			expect(metrics.jobsWithFailures[0].monitorUrl).toBeNull();
+			expect(metrics.jobsWithFailures[0].monitorType).toBeNull();
+			expect(metrics.jobsWithFailures[0].failReason).toBe("boom");
+		});
+	});
+
+	// ── getJobs ────────────────────────────────────────────────────────────────
+
+	describe("getJobs", () => {
+		it("maps scheduler jobs to summaries", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJobs as jest.Mock).mockResolvedValue([
+				{
+					id: "m1",
+					template: "monitor-job",
+					active: true,
+					repeat: 60000,
+					lockedAt: null,
+					runCount: 5,
+					failCount: 1,
+					lastError: "timeout",
+					lastStartedAt: 1000,
+					lastFinishedAt: 1100,
+					lastFailedAt: 900,
+					data: { url: "http://a.com", type: "http", interval: 60000, geoCheckInterval: 300000, isActive: true },
+				},
+			]);
+			const jobs = await queue.getJobs();
+
+			expect(jobs).toHaveLength(1);
+			expect(jobs[0]).toEqual({
+				monitorId: "m1",
+				monitorUrl: "http://a.com",
+				monitorType: "http",
+				monitorInterval: 60000,
+				monitorGeoInterval: 300000,
+				monitorActive: true,
+				active: true,
+				repeat: 60000,
+				lockedAt: null,
+				runCount: 5,
+				failCount: 1,
+				failReason: "timeout",
+				lastRunAt: 1000,
+				lastFinishedAt: 1100,
+				lastRunTook: 100,
+				lastFailedAt: 900,
+			});
+		});
+
+		it("returns null lastRunTook when job is locked", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJobs as jest.Mock).mockResolvedValue([
+				{
+					id: "m1",
+					template: "monitor-job",
+					active: true,
+					lockedAt: 999,
+					runCount: 1,
+					lastStartedAt: 1000,
+					lastFinishedAt: 1100,
+					data: { url: "a", type: "http", interval: 60000 },
+				},
+			]);
+			const jobs = await queue.getJobs();
+			expect(jobs[0].lastRunTook).toBeNull();
+		});
+
+		it("leaves monitor fields null for non-monitor jobs", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.getJobs as jest.Mock).mockResolvedValue([{ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true, data: undefined }]);
+			const jobs = await queue.getJobs();
+			expect(jobs[0].monitorUrl).toBeNull();
+			expect(jobs[0].monitorType).toBeNull();
+			expect(jobs[0].monitorInterval).toBeNull();
+			expect(jobs[0].monitorGeoInterval).toBeNull();
+			expect(jobs[0].monitorActive).toBeNull();
+		});
+	});
+
+	// ── flushQueues ────────────────────────────────────────────────────────────
+
+	describe("flushQueues", () => {
+		it("stops, flushes, and reinitializes", async () => {
+			const { queue, scheduler } = createQueue();
+			const result = await queue.flushQueues();
+			expect(scheduler.stop).toHaveBeenCalled();
+			expect(scheduler.flushJobs).toHaveBeenCalled();
+			expect(result.success).toBe(true);
+		});
+
+		it("returns false when flush fails", async () => {
+			const { queue, scheduler } = createQueue();
+			(scheduler.flushJobs as jest.Mock).mockResolvedValue(false);
+			const result = await queue.flushQueues();
+			expect(result.success).toBe(false);
+		});
+	});
+
+	// ── static create ──────────────────────────────────────────────────────────
+
+	describe("static create", () => {
+		const envSettings = { dbConnectionString: "mongodb://localhost:27017/test_db" } as any;
+
+		it("creates a MongoStore + Scheduler and inits in primary mode", async () => {
+			const logger = createMockLogger();
+			const helper = createQueueHelper();
+			const monitorsRepository = createMonitorsRepo();
+
+			const instance = await LessSimpleQueue.create(logger as any, helper as any, monitorsRepository as any, envSettings, "primary");
+
+			expect(MockMongoStore).toHaveBeenCalledWith(expect.objectContaining({ url: "mongodb://localhost:27017/test_db" }));
+			expect(MockScheduler).toHaveBeenCalled();
+			expect(instance).toBeInstanceOf(LessSimpleQueue);
+			expect(mockSchedulerInstance.start).toHaveBeenCalled();
+			expect(monitorsRepository.findAll).toHaveBeenCalled();
+		});
+
+		it("inits in worker mode without seeding monitors", async () => {
+			const logger = createMockLogger();
+			const helper = createQueueHelper();
+			const monitorsRepository = createMonitorsRepo();
+
+			const instance = await LessSimpleQueue.create(logger as any, helper as any, monitorsRepository as any, envSettings, "worker");
+
+			expect(instance).toBeInstanceOf(LessSimpleQueue);
+			expect(monitorsRepository.findAll).not.toHaveBeenCalled();
+		});
+	});
+});