diff --git a/server/package.json b/server/package.json index ea518a298..559299061 100755 --- a/server/package.json +++ b/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", diff --git a/server/src/config/services.ts b/server/src/config/services.ts index 04966d6e2..ef6233212 100644 --- a/server/src/config/services.ts +++ b/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 => { // 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); diff --git a/server/src/index.ts b/server/src/index.ts index 83698b45c..45c75ac8b 100755 --- a/server/src/index.ts +++ b/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); diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts index 1fa3d3e26..80655ab4c 100644 --- a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts +++ b/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; diff --git a/server/src/service/system/settingsService.ts b/server/src/service/system/settingsService.ts index 97960df3a..dfbf6cfdc 100755 --- a/server/src/service/system/settingsService.ts +++ b/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, }; } diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index 877aef325..3c1fc4ee7 100644 --- a/server/src/shutdown.ts +++ b/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" }); diff --git a/server/src/types/settings.ts b/server/src/types/settings.ts index 7bbf57401..d49bd53dd 100644 --- a/server/src/types/settings.ts +++ b/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; diff --git a/server/src/validation/envValidation.ts b/server/src/validation/envValidation.ts index 7eea70fc5..5280a8036 100644 --- a/server/src/validation/envValidation.ts +++ b/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"), diff --git a/server/test/unit/services/lessSimpleQueue.test.ts b/server/test/unit/services/lessSimpleQueue.test.ts new file mode 100644 index 000000000..b5a79445e --- /dev/null +++ b/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 = {}; + 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) => { + 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 => + ({ + 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(); + }); + }); +});