From 1a0e6d9404303732781f66d4ccc006170b3b919e Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 1 Jun 2026 10:44:05 -0700 Subject: [PATCH 01/39] add queue mode type --- server/src/types/settings.ts | 3 +++ 1 file changed, 3 insertions(+) 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; From fc357fe3aef94ac8780900dc423e8b4a8336eaff Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 1 Jun 2026 10:44:18 -0700 Subject: [PATCH 02/39] add queue mode to env validation --- server/src/validation/envValidation.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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"), From 51672890df61bb031588a36cf077d2bbf625b523 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 1 Jun 2026 10:44:39 -0700 Subject: [PATCH 03/39] add queue type to less simple queue --- .../JobQueues/LessSimpleQueue.ts | 35 ++++++++++++------- server/src/service/system/settingsService.ts | 4 ++- 2 files changed, 25 insertions(+), 14 deletions(-) diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts index 1fa3d3e26..00a8db33f 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,22 +116,30 @@ 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()); - const monitors = await this.monitorsRepository.findAll(); - if (!monitors) { + + // 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(); for (const monitor of monitors) { const randomOffset = Math.floor(Math.random() * 100); setTimeout(() => { 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, }; } From e1558d35b0e2d190817f1dd6889c0b7c9924199c Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 1 Jun 2026 10:44:53 -0700 Subject: [PATCH 04/39] use queue type on config --- server/src/config/services.ts | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/server/src/config/services.ts b/server/src/config/services.ts index 04966d6e2..ed8e3a83a 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -123,7 +123,9 @@ 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 { QueueModes, type QueueMode } from "@/types/settings.js"; +import { env } from "process"; 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 => { // Create DB @@ -345,10 +349,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); From f74231cf33a649109c0e9434bcb6b82e9ed4bb26 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 1 Jun 2026 10:45:09 -0700 Subject: [PATCH 05/39] early return if worker mode --- server/src/index.ts | 29 +++++++++++++++++++---------- server/src/shutdown.ts | 4 ++-- 2 files changed, 21 insertions(+), 12 deletions(-) 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/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" }); From 4df08bfa7905fd8e97cc679ab14be1615ef91772 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 1 Jun 2026 10:45:20 -0700 Subject: [PATCH 06/39] add worker scripts --- server/package.json | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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", From e37fbf2a5c065941811a00d856a7c5a4409bebd2 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 1 Jun 2026 11:11:36 -0700 Subject: [PATCH 07/39] replace empty monitor check --- server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts index 00a8db33f..80655ab4c 100644 --- a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts @@ -140,6 +140,9 @@ export class LessSimpleQueue implements IJobQueue { } const monitors = await this.monitorsRepository.findAll(); + if (!monitors) { + return true; + } for (const monitor of monitors) { const randomOffset = Math.floor(Math.random() * 100); setTimeout(() => { From 53ed3b34655f2a4c7a5eb4c26ad80dc549e4ab49 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 1 Jun 2026 11:17:50 -0700 Subject: [PATCH 08/39] remove unused imports --- server/src/config/services.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/server/src/config/services.ts b/server/src/config/services.ts index ed8e3a83a..ef6233212 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -124,8 +124,7 @@ import { } from "@/repositories/index.js"; import { ILogger } from "@/utils/logger.js"; import { AppError } from "@/utils/AppError.js"; -import { QueueModes, type QueueMode } from "@/types/settings.js"; -import { env } from "process"; +import { type QueueMode } from "@/types/settings.js"; export type InitializedServices = { settingsService: ISettingsService; From 89c4f4005e4266b89365a781a7b6b9fcf6ff2a90 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 1 Jun 2026 11:21:49 -0700 Subject: [PATCH 09/39] add lessSimpleQueue tests --- .../unit/services/lessSimpleQueue.test.ts | 652 ++++++++++++++++++ 1 file changed, 652 insertions(+) create mode 100644 server/test/unit/services/lessSimpleQueue.test.ts 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(); + }); + }); +}); From 7df3ee6ecd1a61a600879efcba2a3112b0ebd416 Mon Sep 17 00:00:00 2001 From: Br0wnHammer Date: Tue, 2 Jun 2026 20:27:12 +0530 Subject: [PATCH 10/39] feat(status-page): add wizard step metadata and helpers to validation schema --- client/src/Validation/statusPage.ts | 56 ++++++++++++++++++++++++----- 1 file changed, 47 insertions(+), 9 deletions(-) diff --git a/client/src/Validation/statusPage.ts b/client/src/Validation/statusPage.ts index 8b41b7780..bd4e95c01 100644 --- a/client/src/Validation/statusPage.ts +++ b/client/src/Validation/statusPage.ts @@ -1,6 +1,14 @@ import { z } from "zod"; +import type { FieldPath } from "react-hook-form"; import { STATUS_PAGE_THEMES, STATUS_PAGE_THEME_MODES } from "@/Types/StatusPage"; +// Wizard step a field is validated on. Attached inline to each field below so +// the grouping lives next to the field definition; unannotated fields default +// to step 0. Read via `statusPageStepFieldsFor`. Mirrors the monitor schema's +// `monitorStepRegistry` so the create-status-page wizard can't drift from the +// field definitions. +export const statusPageStepRegistry = z.registry<{ step: number }>(); + export const statusPageSchema = z.object({ companyName: z .string() @@ -18,23 +26,53 @@ export const statusPageSchema = z.object({ type: z .array(z.enum(["uptime", "infrastructure"])) .min(1, "At least one type is required"), - color: z.string().min(1, "Color is required"), + color: z.string().min(1, "Color is required").register(statusPageStepRegistry, { + step: 1, + }), monitors: z.array(z.string()).min(1, "At least one monitor is required"), isPublished: z.boolean(), - showCharts: z.boolean(), - showUptimePercentage: z.boolean(), - showAdminLoginLink: z.boolean(), - showInfrastructure: z.boolean(), - customCSS: z.string().optional(), - theme: z.enum(STATUS_PAGE_THEMES).optional(), - themeMode: z.enum(STATUS_PAGE_THEME_MODES).optional(), + showCharts: z.boolean().register(statusPageStepRegistry, { step: 1 }), + showUptimePercentage: z.boolean().register(statusPageStepRegistry, { step: 1 }), + showAdminLoginLink: z.boolean().register(statusPageStepRegistry, { step: 1 }), + showInfrastructure: z.boolean().register(statusPageStepRegistry, { step: 1 }), + customCSS: z.string().optional().register(statusPageStepRegistry, { step: 1 }), + theme: z + .enum(STATUS_PAGE_THEMES) + .optional() + .register(statusPageStepRegistry, { step: 1 }), + themeMode: z + .enum(STATUS_PAGE_THEME_MODES) + .optional() + .register(statusPageStepRegistry, { step: 1 }), logo: z .object({ data: z.string(), contentType: z.string(), }) .nullable() - .optional(), + .optional() + .register(statusPageStepRegistry, { step: 1 }), }); export type StatusPageFormData = z.infer; + +// The step each field is validated on, from the per-field +// `statusPageStepRegistry` metadata (unannotated fields default to step 0). +const fieldSteps = (): [name: FieldPath, step: number][] => + Object.entries(statusPageSchema.shape).map(([name, field]) => [ + name as FieldPath, + statusPageStepRegistry.get(field)?.step ?? 0, + ]); + +// The fields that belong to a given wizard step. Derived from the schema so it +// can't drift from the field definitions. +export const statusPageStepFieldsFor = ( + step: number +): FieldPath[] => + fieldSteps() + .filter(([, fieldStep]) => fieldStep === step) + .map(([name]) => name); + +// Number of wizard steps, derived from the highest step its fields declare. +export const statusPageStepCount = (): number => + Math.max(0, ...fieldSteps().map(([, step]) => step)) + 1; From 76179297007b5dee84ac82828af510cd2e938863 Mon Sep 17 00:00:00 2001 From: Br0wnHammer Date: Tue, 2 Jun 2026 20:31:01 +0530 Subject: [PATCH 11/39] feat(status-page): add step-wizard progress indicator to create flow --- client/src/Pages/StatusPage/Create/index.tsx | 663 ++++++++++--------- client/src/Validation/statusPage.ts | 4 +- 2 files changed, 362 insertions(+), 305 deletions(-) diff --git a/client/src/Pages/StatusPage/Create/index.tsx b/client/src/Pages/StatusPage/Create/index.tsx index 5ef5095ea..c9bc04838 100644 --- a/client/src/Pages/StatusPage/Create/index.tsx +++ b/client/src/Pages/StatusPage/Create/index.tsx @@ -1,4 +1,4 @@ -import { BasePage, ConfigBox } from "@/Components/design-elements"; +import { BasePage, ConfigBox, StepProgress } from "@/Components/design-elements"; import Stack from "@mui/material/Stack"; import { logger } from "@/Utils/logger"; import { SPACING, LAYOUT } from "@/Utils/Theme/constants"; @@ -26,6 +26,7 @@ import { useForm, Controller } from "react-hook-form"; import { zodResolver } from "@hookform/resolvers/zod"; import { useStatusPageForm } from "@/Hooks/useStatusPageForm"; import type { StatusPageFormData } from "@/Validation/statusPage"; +import { statusPageStepCount, statusPageStepFieldsFor } from "@/Validation/statusPage"; import { useGet, usePost, usePut, useDelete } from "@/Hooks/UseApi"; import type { Monitor } from "@/Types/Monitor"; import type { MonitorDisplayType, StatusPageResponse } from "@/Types/StatusPage"; @@ -86,13 +87,27 @@ const CreateStatusPage = () => { defaultValues: defaults, }); - const { control, reset, handleSubmit } = form; + const { control, reset, handleSubmit, trigger } = form; // Reset form when defaults change (from fetched data) useEffect(() => { reset(defaults); }, [defaults, reset]); + // Multi-step wizard, create only. The configure (edit) flow keeps the flat + // form, where showStep() is always true. + const [currentStep, setCurrentStep] = useState(0); + const showStep = (step: number) => !isCreate || currentStep === step; + const totalSteps = statusPageStepCount(); + + const handleNext = async () => { + const isValid = await trigger(statusPageStepFieldsFor(currentStep)); + if (isValid) { + setCurrentStep((step) => step + 1); + } + }; + const handleBack = () => setCurrentStep((step) => step - 1); + const watchedMonitorIds: string[] = form.watch("monitors") ?? []; const computedTypes: MonitorDisplayType[] = useMemo(() => { const selectedMonitors = (watchedMonitorIds ?? []) @@ -200,315 +215,334 @@ const CreateStatusPage = () => { onSubmit={handleSubmit(onSubmit, onError)} > {!isCreate && } - - ( - field.onChange(e.target.checked)} - /> - )} - /> - - {t("pages.statusPages.form.access.option.published.name")} - - - } - /> - - ( - - )} - /> - ( - - )} - /> - - } - /> - { - const selectedMonitors = field.value - .map((id: string) => monitors.find((m) => m.id === id)) - .filter((m): m is Monitor => m !== undefined); - - const handleDragEnd = (result: DropResult) => { - if (!result.destination) return; - const reordered = Array.from(field.value); - const [removed] = reordered.splice(result.source.index, 1); - reordered.splice(result.destination.index, 0, removed); - field.onChange(reordered); - }; - - return ( - - option.name} - value={selectedMonitors} - onChange={(_, newValue) => { - field.onChange(newValue.map((m: Monitor) => m.id)); - }} - fieldLabel={t( - "pages.statusPages.form.monitors.option.monitors.label" - )} - renderInput={(params) => ( - - )} - /> - {selectedMonitors.length > 0 && ( - - - {(provided) => ( - - {selectedMonitors.map((monitor, index) => ( - - {(provided) => ( - - - {`${monitor.name} (${getMonitorTypeLabel(monitor.type, t)})`} - { - field.onChange( - field.value.filter( - (id: string) => id !== monitor.id - ) - ); - }} - aria-label="Remove monitor" - > - - - - )} - - ))} - {provided.placeholder} - - )} - - - )} - - ); - }} - /> - } - /> - ( - option.name} - value={ - timezones.find((tz: TimezoneOption) => tz._id === field.value) ?? null - } - onChange={(_, newValue: TimezoneOption | null) => { - field.onChange(newValue?._id ?? ""); - }} - fieldLabel={t("pages.statusPages.form.timezone.option.timezone.label")} - renderInput={(params) => ( - + )} + {showStep(0) && ( + + ( + field.onChange(e.target.checked)} /> )} /> - )} - /> - } - /> - - + + {t("pages.statusPages.form.access.option.published.name")} + + + } + /> + )} + {showStep(0) && ( + ( - { - if (file) { - field.onChange({ - data: file.src, - contentType: file.file.type, - }); - } else { - field.onChange(null); - } - }} + render={({ field, fieldState }) => ( + + )} + /> + ( + )} /> + } + /> + )} + {showStep(0) && ( + ( - - )} - /> - - } - /> - ( - ( - - )} - /> - )} - /> - } - /> - - ( - { + const selectedMonitors = field.value + .map((id: string) => monitors.find((m) => m.id === id)) + .filter((m): m is Monitor => m !== undefined); + + const handleDragEnd = (result: DropResult) => { + if (!result.destination) return; + const reordered = Array.from(field.value); + const [removed] = reordered.splice(result.source.index, 1); + reordered.splice(result.destination.index, 0, removed); + field.onChange(reordered); + }; + + return ( + + option.name} + value={selectedMonitors} + onChange={(_, newValue) => { + field.onChange(newValue.map((m: Monitor) => m.id)); + }} + fieldLabel={t( + "pages.statusPages.form.monitors.option.monitors.label" + )} + renderInput={(params) => ( + + )} /> - } - label={t("pages.statusPages.form.features.option.showCharts.label")} - /> - )} + {selectedMonitors.length > 0 && ( + + + {(provided) => ( + + {selectedMonitors.map((monitor, index) => ( + + {(provided) => ( + + + {`${monitor.name} (${getMonitorTypeLabel(monitor.type, t)})`} + { + field.onChange( + field.value.filter( + (id: string) => id !== monitor.id + ) + ); + }} + aria-label="Remove monitor" + > + + + + )} + + ))} + {provided.placeholder} + + )} + + + )} + + ); + }} /> + } + /> + )} + {showStep(0) && ( + ( - + option.name} + value={ + timezones.find((tz: TimezoneOption) => tz._id === field.value) ?? null } - label={t( - "pages.statusPages.form.features.option.showInfrastructure.label" + onChange={(_, newValue: TimezoneOption | null) => { + field.onChange(newValue?._id ?? ""); + }} + fieldLabel={t("pages.statusPages.form.timezone.option.timezone.label")} + renderInput={(params) => ( + )} /> )} /> - {/* + )} + {showStep(1) && ( + + + ( + { + if (file) { + field.onChange({ + data: file.src, + contentType: file.file.type, + }); + } else { + field.onChange(null); + } + }} + /> + )} + /> + + ( + + )} + /> + + } + /> + )} + {showStep(1) && ( + ( + ( + + )} + /> + )} + /> + } + /> + )} + {showStep(1) && ( + + ( + + } + label={t("pages.statusPages.form.features.option.showCharts.label")} + /> + )} + /> + ( + + } + label={t( + "pages.statusPages.form.features.option.showInfrastructure.label" + )} + /> + )} + /> + {/* ( @@ -542,21 +576,46 @@ const CreateStatusPage = () => { /> )} /> */} - - } - /> + + } + /> + )} - + {isCreate && ( + + )} + {isCreate && currentStep < totalSteps - 1 ? ( + + ) : ( + + )} , step: number][] => // The fields that belong to a given wizard step. Derived from the schema so it // can't drift from the field definitions. -export const statusPageStepFieldsFor = ( - step: number -): FieldPath[] => +export const statusPageStepFieldsFor = (step: number): FieldPath[] => fieldSteps() .filter(([, fieldStep]) => fieldStep === step) .map(([name]) => name); From 60356a95fe0060f014adf1941315646e70584ae7 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:44:23 -0700 Subject: [PATCH 12/39] bump less-simple-scheduler --- server/package-lock.json | 8 ++++---- server/package.json | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/server/package-lock.json b/server/package-lock.json index 131589613..be543815d 100644 --- a/server/package-lock.json +++ b/server/package-lock.json @@ -31,7 +31,7 @@ "jmespath": "^0.16.0", "jsdom": "^26.1.0", "jsonwebtoken": "9.0.2", - "less-simple-scheduler": "2.2.0", + "less-simple-scheduler": "2.3.0", "mailersend": "^2.2.0", "mjml": "^5.0.0-alpha.4", "mongoose": "^8.3.3", @@ -10417,9 +10417,9 @@ "license": "MIT" }, "node_modules/less-simple-scheduler": { - "version": "2.2.0", - "resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.2.0.tgz", - "integrity": "sha512-0bbFohlkcWZY2wfkGfDiYEbDxkshj0RuskkL0RLUDMn1Y3WjoPqy29PAnKNtnHG+DQqn0hMjZ0e3hVFJ4k76oA==", + "version": "2.3.0", + "resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.3.0.tgz", + "integrity": "sha512-HL8f8YCcNav03oNTWFQxkWmoqtEzsIXNOgLcVmOEAgFxsALukOk4bDlem6SlZmjVAsip8Y50QUu6ZsblxqQYmA==", "license": "ISC", "dependencies": { "mongodb": "7.2.0" diff --git a/server/package.json b/server/package.json index 559299061..c9b85e17c 100755 --- a/server/package.json +++ b/server/package.json @@ -53,7 +53,7 @@ "jmespath": "^0.16.0", "jsdom": "^26.1.0", "jsonwebtoken": "9.0.2", - "less-simple-scheduler": "2.2.0", + "less-simple-scheduler": "2.3.0", "mailersend": "^2.2.0", "mjml": "^5.0.0-alpha.4", "mongoose": "^8.3.3", From 8b2506cc3d8200d0481dddcf24cf4726060fdc7f Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:44:49 -0700 Subject: [PATCH 13/39] add queue worker repository --- server/src/repositories/index.ts | 3 ++ .../queue-workers/IQueueWorkersRepository.ts | 7 ++++ .../MongoQueueWorkersRepository.ts | 33 +++++++++++++++++++ 3 files changed, 43 insertions(+) create mode 100644 server/src/repositories/queue-workers/IQueueWorkersRepository.ts create mode 100644 server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts diff --git a/server/src/repositories/index.ts b/server/src/repositories/index.ts index f45768b02..d36d9d189 100644 --- a/server/src/repositories/index.ts +++ b/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"; diff --git a/server/src/repositories/queue-workers/IQueueWorkersRepository.ts b/server/src/repositories/queue-workers/IQueueWorkersRepository.ts new file mode 100644 index 000000000..b1f7e76b1 --- /dev/null +++ b/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; + findRecent(maxAgeMs: number): Promise; + deleteById(workerId: string): Promise; +} diff --git a/server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts b/server/src/repositories/queue-workers/MongoQueueWorkersRepository.ts new file mode 100644 index 000000000..b8155ea1d --- /dev/null +++ b/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 => { + await QueueWorkerModel.updateOne({ _id: workerId }, { $set: { mode, lastSeenAt: new Date() } }, { upsert: true }); + }; + + findRecent = async (maxAgeMs: number): Promise => { + 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 => { + await QueueWorkerModel.deleteOne({ _id: workerId }); + }; +} + +export default MongoQueueWorkersRepository; From cffe3f57facb9fbcfeb6a210fe2ab145fe0014c4 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:45:37 -0700 Subject: [PATCH 14/39] add models --- server/src/db/models/QueueWorker.ts | 28 ++++++++++++++++++++++++++++ server/src/db/models/index.ts | 3 +++ 2 files changed, 31 insertions(+) create mode 100644 server/src/db/models/QueueWorker.ts diff --git a/server/src/db/models/QueueWorker.ts b/server/src/db/models/QueueWorker.ts new file mode 100644 index 000000000..76485f56c --- /dev/null +++ b/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( + { + _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("QueueWorker", QueueWorkerSchema); + +export type { QueueWorkerDocument }; +export { QueueWorkerModel }; +export default QueueWorkerModel; diff --git a/server/src/db/models/index.ts b/server/src/db/models/index.ts index f0647f9eb..b825ca503 100644 --- a/server/src/db/models/index.ts +++ b/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"; From b2526a5be92b5fa02d94507d9d27a3593ef54eec Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:45:56 -0700 Subject: [PATCH 15/39] init queue worker repo --- server/src/config/services.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/server/src/config/services.ts b/server/src/config/services.ts index ef6233212..514bd04d5 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -107,6 +107,7 @@ import { MongoTeamsRepository, MongoMaintenanceWindowsRepository, MongoSettingsRepository, + MongoQueueWorkersRepository, IMonitorsRepository, IChecksRepository, IGeoChecksRepository, @@ -248,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); @@ -360,7 +362,7 @@ export const initializeServices = async ({ let jobQueue: IJobQueue; switch (envSettings.queueType) { case "lessSimpleQueue": - jobQueue = await LessSimpleQueue.create(logger, queueHelper, monitorsRepository, envSettings, queueMode); + jobQueue = await LessSimpleQueue.create(logger, queueHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode); break; default: jobQueue = await SuperSimpleQueue.create(logger, queueHelper, monitorsRepository); From c9eacff71f81e5a23b9c870ae567b9852073a870 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:46:23 -0700 Subject: [PATCH 16/39] add queue worker type --- server/src/types/settings.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/server/src/types/settings.ts b/server/src/types/settings.ts index d49bd53dd..4a4ff02a2 100644 --- a/server/src/types/settings.ts +++ b/server/src/types/settings.ts @@ -7,6 +7,12 @@ 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; From cfed235d1f216e6364096a125d8acb392cc706f1 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:47:00 -0700 Subject: [PATCH 17/39] add worker stats --- .../infrastructure/JobQueues/IJobQueue.ts | 4 + .../JobQueues/LessSimpleQueue.ts | 74 +++++++++++++++++-- .../JobQueues/SuperSimpleQueue.ts | 3 + 3 files changed, 76 insertions(+), 5 deletions(-) diff --git a/server/src/service/infrastructure/JobQueues/IJobQueue.ts b/server/src/service/infrastructure/JobQueues/IJobQueue.ts index a97c8eb6a..371a7c74a 100644 --- a/server/src/service/infrastructure/JobQueues/IJobQueue.ts +++ b/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; diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts index 80655ab4c..cedf29901 100644 --- a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts @@ -1,4 +1,4 @@ -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 { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js"; @@ -8,6 +8,12 @@ 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. The +// model's TTL index uses the same window to garbage-collect stale records. +// Heartbeats are driven by the scheduler's `scheduler:heartbeat` event (emitted +// every lockMs/3, so ~5s with the default lockMs), comfortably inside this TTL. +const WORKER_TTL_MS = 30_000; + // For discriminating job types type MonitorJob = Omit & { data: Monitor }; const MONITOR_TEMPLATES = new Set(["monitor-job", "geo-check-job"]); @@ -19,12 +25,21 @@ export class LessSimpleQueue implements IJobQueue { private logger: ILogger; private helper: IQueueHelper; private monitorsRepository: IMonitorsRepository; + private workersRepository: IQueueWorkersRepository; private readonly scheduler: Scheduler; + 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 + ) { this.logger = logger; this.helper = helper; this.monitorsRepository = monitorsRepository; + this.workersRepository = workersRepository; this.scheduler = scheduler; this.registerListeners(); } @@ -108,19 +123,45 @@ export class LessSimpleQueue implements IJobQueue { }); }; - static async create(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, envSettings: EnvConfig, queueMode: QueueMode) { + 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, }); - const instance = new LessSimpleQueue(logger, helper, monitorsRepository, scheduler); + const instance = new LessSimpleQueue(logger, helper, monitorsRepository, workersRepository, scheduler); await instance.init(queueMode); return instance; } + private startHeartbeat = (queueMode: QueueMode) => { + const beat = ( + workerId: string // This is the heartbeat function that registers workers + ) => + this.workersRepository.upsert(workerId, queueMode).catch((error: unknown) => { + this.logger.warn({ + message: `Worker heartbeat failed: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + method: "startHeartbeat", + }); + }); + beat(this.scheduler.workerId); // Beat immediately to register + if (this.heartbeatListener) { + this.scheduler.off("scheduler:heartbeat", this.heartbeatListener); // remove old listeners + } + this.heartbeatListener = beat; /// Keep a reference to the listener + this.scheduler.on("scheduler:heartbeat", beat); // beat on every heartbeat + }; + init = async (queueMode: QueueMode = "primary") => { try { await this.scheduler.start(); @@ -129,6 +170,9 @@ export class LessSimpleQueue implements IJobQueue { 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({ @@ -284,11 +328,26 @@ 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 now = Date.now(); + const metrics = jobs.reduce( (acc, job) => { const runCount = job.runCount ?? 0; @@ -302,7 +361,8 @@ export class LessSimpleQueue implements IJobQueue { acc.failingJobs++; } - if (job.lockedAt) { + // A job currently held by a live lock is in flight on some worker + if (job.lockedBy && job.lockedUntil && job.lockedUntil > now) { acc.activeJobs++; } @@ -326,8 +386,10 @@ export class LessSimpleQueue implements IJobQueue { jobsWithFailures: [], totalRuns: 0, totalFailures: 0, + workers: [], } ); + metrics.workers = await this.workersRepository.findRecent(WORKER_TTL_MS); return metrics; }; @@ -345,6 +407,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, diff --git a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts index df265cf71..353a7ec7c 100644 --- a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts @@ -354,6 +354,7 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { }>, totalRuns: 0, totalFailures: 0, + workers: [], } ); return metrics; @@ -371,6 +372,8 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { monitorActive: job.data?.isActive ?? null, active: job.active, repeat: job.repeat ?? null, + lockedBy: null, + lockedUntil: null, lockedAt: job.lockedAt ?? null, runCount: job.runCount ?? 0, failCount: job.failCount ?? 0, From 599c5995881dd028422a74d8c2d6e7374af61851 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 09:47:13 -0700 Subject: [PATCH 18/39] add tests --- .../unit/services/lessSimpleQueue.test.ts | 150 +++++++++++++++++- .../unit/services/superSimpleQueue.test.ts | 4 +- 2 files changed, 147 insertions(+), 7 deletions(-) diff --git a/server/test/unit/services/lessSimpleQueue.test.ts b/server/test/unit/services/lessSimpleQueue.test.ts index b5a79445e..1b455ef1b 100644 --- a/server/test/unit/services/lessSimpleQueue.test.ts +++ b/server/test/unit/services/lessSimpleQueue.test.ts @@ -12,6 +12,9 @@ const createScheduler = () => { listeners[event] = listeners[event] || []; listeners[event].push(cb); }), + off: jest.fn((event: string, cb: Function) => { + listeners[event] = (listeners[event] || []).filter((fn) => fn !== cb); + }), emit: (event: string, ...args: unknown[]) => { (listeners[event] || []).forEach((cb) => cb(...args)); }, @@ -51,18 +54,34 @@ const createMonitorsRepo = () => ({ findAll: jest.fn().mockResolvedValue([]), }); +const createWorkersRepo = () => ({ + upsert: jest.fn().mockResolvedValue(undefined), + findRecent: jest.fn().mockResolvedValue([]), + deleteById: jest.fn().mockResolvedValue(undefined), +}); + const createQueue = (overrides?: Record) => { const logger = createMockLogger(); const helper = createQueueHelper(); const monitorsRepository = createMonitorsRepo(); + const workersRepository = createWorkersRepo(); const scheduler = createScheduler(); - const defaults = { logger, helper, monitorsRepository, scheduler, ...overrides }; + const defaults = { logger, helper, monitorsRepository, workersRepository, scheduler, ...overrides }; - const queue = new LessSimpleQueue(defaults.logger as any, defaults.helper as any, defaults.monitorsRepository as any, defaults.scheduler as any); + const queue = new LessSimpleQueue( + defaults.logger as any, + defaults.helper as any, + defaults.monitorsRepository as any, + defaults.workersRepository as any, + defaults.scheduler as any + ); return { queue, ...defaults }; }; +// A lockedUntil far enough ahead that it is always > Date.now() at assertion time. +const FAR_FUTURE = 4_000_000_000_000; // year 2096 + const makeMonitor = (overrides?: Partial): Monitor => ({ id: "mon-1", @@ -112,7 +131,8 @@ describe("LessSimpleQueue", () => { (monitorsRepository.findAll as jest.Mock).mockResolvedValue(monitors); await queue.init(); - jest.runAllTimers(); + // advance past the <100ms seed offsets + jest.advanceTimersByTime(100); expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "m1", template: "monitor-job" })); expect(scheduler.addJob).toHaveBeenCalledWith(expect.objectContaining({ id: "m2", template: "monitor-job" })); @@ -454,7 +474,7 @@ describe("LessSimpleQueue", () => { 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 }); + expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0, workers: [] }); }); it("aggregates metrics from jobs", async () => { @@ -478,6 +498,9 @@ describe("LessSimpleQueue", () => { failCount: 0, lastFailedAt: 0, lastFinishedAt: 50, + // live lock held by a worker + lockedBy: "host-1:1234:abcd", + lockedUntil: FAR_FUTURE, lockedAt: 999, data: { url: "http://b.com", type: "ping" }, }, @@ -526,6 +549,56 @@ describe("LessSimpleQueue", () => { }); }); + // ── getMetrics: activeJobs (in-flight) ─────────────────────────────────────── + + describe("getMetrics - activeJobs", () => { + it("counts jobs held by a live lock as active", async () => { + const { queue, scheduler } = createQueue(); + (scheduler.getJobs as jest.Mock).mockResolvedValue([ + { id: "m1", template: "monitor-job", lockedBy: "worker-a", lockedUntil: FAR_FUTURE, lockedAt: 100, data: { url: "a" } }, + { id: "m2", template: "monitor-job", lockedBy: "worker-b", lockedUntil: FAR_FUTURE, lockedAt: 250, data: { url: "b" } }, + ]); + const metrics = await queue.getMetrics(); + expect(metrics.activeJobs).toBe(2); + }); + + it("does not count unlocked jobs (lockedBy null)", async () => { + const { queue, scheduler } = createQueue(); + (scheduler.getJobs as jest.Mock).mockResolvedValue([ + { id: "m1", template: "monitor-job", lockedBy: null, lockedUntil: null, lockedAt: null, data: { url: "a" } }, + ]); + const metrics = await queue.getMetrics(); + expect(metrics.activeJobs).toBe(0); + }); + + it("does not count jobs whose lock has expired (lockedUntil <= now)", async () => { + const { queue, scheduler } = createQueue(); + (scheduler.getJobs as jest.Mock).mockResolvedValue([ + { id: "m1", template: "monitor-job", lockedBy: "worker-a", lockedUntil: 1, lockedAt: 1, data: { url: "a" } }, + ]); + const metrics = await queue.getMetrics(); + expect(metrics.activeJobs).toBe(0); + }); + }); + + // ── getMetrics: worker registry ────────────────────────────────────────────── + + describe("getMetrics - worker registry", () => { + it("returns the alive workers from the registry", async () => { + const { queue, workersRepository } = createQueue(); + const alive = [ + { workerId: "host:1:aaa", mode: "primary", lastSeenAt: 1000 }, + { workerId: "host:2:bbb", mode: "worker", lastSeenAt: 2000 }, + ]; + (workersRepository.findRecent as jest.Mock).mockResolvedValue(alive); + + const metrics = await queue.getMetrics(); + + expect(workersRepository.findRecent).toHaveBeenCalledWith(30000); + expect(metrics.workers).toEqual(alive); + }); + }); + // ── getJobs ──────────────────────────────────────────────────────────────── describe("getJobs", () => { @@ -559,6 +632,8 @@ describe("LessSimpleQueue", () => { monitorActive: true, active: true, repeat: 60000, + lockedBy: null, + lockedUntil: null, lockedAt: null, runCount: 5, failCount: 1, @@ -628,8 +703,16 @@ describe("LessSimpleQueue", () => { const logger = createMockLogger(); const helper = createQueueHelper(); const monitorsRepository = createMonitorsRepo(); + const workersRepository = createWorkersRepo(); - const instance = await LessSimpleQueue.create(logger as any, helper as any, monitorsRepository as any, envSettings, "primary"); + const instance = await LessSimpleQueue.create( + logger as any, + helper as any, + monitorsRepository as any, + workersRepository as any, + envSettings, + "primary" + ); expect(MockMongoStore).toHaveBeenCalledWith(expect.objectContaining({ url: "mongodb://localhost:27017/test_db" })); expect(MockScheduler).toHaveBeenCalled(); @@ -642,11 +725,66 @@ describe("LessSimpleQueue", () => { const logger = createMockLogger(); const helper = createQueueHelper(); const monitorsRepository = createMonitorsRepo(); + const workersRepository = createWorkersRepo(); - const instance = await LessSimpleQueue.create(logger as any, helper as any, monitorsRepository as any, envSettings, "worker"); + const instance = await LessSimpleQueue.create( + logger as any, + helper as any, + monitorsRepository as any, + workersRepository as any, + envSettings, + "worker" + ); expect(instance).toBeInstanceOf(LessSimpleQueue); expect(monitorsRepository.findAll).not.toHaveBeenCalled(); }); }); + + // ── worker heartbeat / deregistration ──────────────────────────────────────── + + describe("worker heartbeat", () => { + it("writes a heartbeat on init with the workerId and mode (primary)", async () => { + const { queue, workersRepository } = createQueue(); + await queue.init("primary"); + expect(workersRepository.upsert).toHaveBeenCalledWith("host-1:1234:abcd", "primary"); + }); + + it("writes a heartbeat on init with the workerId and mode (worker)", async () => { + const { queue, workersRepository } = createQueue(); + await queue.init("worker"); + expect(workersRepository.upsert).toHaveBeenCalledWith("host-1:1234:abcd", "worker"); + }); + + it("beats again on each scheduler:heartbeat event", async () => { + const { queue, scheduler, workersRepository } = createQueue(); + await queue.init("worker"); + (workersRepository.upsert as jest.Mock).mockClear(); + + scheduler.emit("scheduler:heartbeat", "host-1:1234:abcd"); + scheduler.emit("scheduler:heartbeat", "host-1:1234:abcd"); + + expect(workersRepository.upsert).toHaveBeenCalledTimes(2); + expect(workersRepository.upsert).toHaveBeenCalledWith("host-1:1234:abcd", "worker"); + }); + + it("stops beating after shutdown", async () => { + const { queue, scheduler, workersRepository } = createQueue(); + await queue.init("worker"); + await queue.shutdown(); + (workersRepository.upsert as jest.Mock).mockClear(); + + scheduler.emit("scheduler:heartbeat", "host-1:1234:abcd"); + + expect(workersRepository.upsert).not.toHaveBeenCalled(); + }); + + it("deregisters the worker and stops the scheduler on shutdown", async () => { + const { queue, scheduler, workersRepository } = createQueue(); + await queue.init("worker"); + await queue.shutdown(); + expect(workersRepository.deleteById).toHaveBeenCalledWith("host-1:1234:abcd"); + expect(scheduler.stop).toHaveBeenCalled(); + }); + }); }); diff --git a/server/test/unit/services/superSimpleQueue.test.ts b/server/test/unit/services/superSimpleQueue.test.ts index 660d0b8f7..970706199 100644 --- a/server/test/unit/services/superSimpleQueue.test.ts +++ b/server/test/unit/services/superSimpleQueue.test.ts @@ -489,7 +489,7 @@ describe("SuperSimpleQueue", () => { 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 }); + expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0, workers: [] }); }); it("aggregates metrics from jobs", async () => { @@ -580,6 +580,8 @@ describe("SuperSimpleQueue", () => { monitorActive: true, active: true, repeat: 60000, + lockedBy: null, + lockedUntil: null, lockedAt: null, runCount: 5, failCount: 1, From caf8f22fec0523a06cf9b5fefed4cb902085a1c4 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 10:59:14 -0700 Subject: [PATCH 19/39] add worker table --- client/src/Pages/Logs/TabQueue.tsx | 34 ++++++- .../src/Pages/Logs/components/TableJobs.tsx | 21 +++++ .../Pages/Logs/components/TableWorkers.tsx | 94 +++++++++++++++++++ client/src/Pages/Logs/index.tsx | 2 +- client/src/Types/Queue.ts | 22 ++++- client/src/locales/en.json | 8 ++ 6 files changed, 174 insertions(+), 7 deletions(-) create mode 100644 client/src/Pages/Logs/components/TableWorkers.tsx diff --git a/client/src/Pages/Logs/TabQueue.tsx b/client/src/Pages/Logs/TabQueue.tsx index a39250be0..bb3aac368 100644 --- a/client/src/Pages/Logs/TabQueue.tsx +++ b/client/src/Pages/Logs/TabQueue.tsx @@ -2,7 +2,7 @@ import Stack from "@mui/material/Stack"; import Typography from "@mui/material/Typography"; import Box from "@mui/material/Box"; import { TableJobs, TableFailedJobs } from "./components/TableJobs"; - +import { TableWorkers } from "./components/TableWorkers"; import { useTheme } from "@mui/material"; import { useTranslation, Trans } from "react-i18next"; import { useGet, usePost } from "@/Hooks/UseApi"; @@ -25,6 +25,10 @@ export const TabQueue = () => { const { post, loading: isFlushing } = usePost(); const jobs = queueData?.jobs ?? []; + const workers = (queueData?.metrics?.workers ?? []).map((worker, idx) => ({ + ...worker, + id: idx, + })); const metrics = queueData?.metrics ?? null; const hasNeverRun = !isLoading && metrics !== null && (metrics.totalRuns ?? 0) === 0; const noQueueData = !isLoading && metrics === null && jobs.length === 0; @@ -46,6 +50,34 @@ export const TabQueue = () => { return ( + + + + {t("pages.logs.workers")} + + + + ), + }} + /> + + + + { ); }, }, + { + id: "lockedBy", + content: t("pages.logs.table.headers.lockedBy"), + render: (row) => { + const lockedBy = row.lockedBy; + if (!lockedBy) { + return -; + } + + const workerId = lockedBy.split(":")[2]; + const shortWorkerId = workerId.slice(workerId.length - 8); + return ( + + {`...${shortWorkerId}`} + + ); + }, + }, ]; return ( diff --git a/client/src/Pages/Logs/components/TableWorkers.tsx b/client/src/Pages/Logs/components/TableWorkers.tsx new file mode 100644 index 000000000..5d1f44860 --- /dev/null +++ b/client/src/Pages/Logs/components/TableWorkers.tsx @@ -0,0 +1,94 @@ +import { Table } from "@/Components/design-elements"; +import { Typography, useTheme } from "@mui/material"; +import { formatTimestamp } from "@/Utils/TimeUtils"; + +import { useTranslation } from "react-i18next"; +import type { QueueWorker } from "@/Types/Queue"; +import type { Header } from "@/Components/design-elements"; + +type QueueWorkerWithId = QueueWorker & { id: number }; + +interface TableWorkersProps { + queueWorkers: QueueWorkerWithId[]; +} + +export const TableWorkers = ({ queueWorkers }: TableWorkersProps) => { + const { t } = useTranslation(); + const theme = useTheme(); + + const cellSx = { + whiteSpace: "nowrap" as const, + overflow: "hidden", + textOverflow: "ellipsis", + maxWidth: 220, + }; + + const headers: Header[] = [ + { + id: "host", + content: t("pages.logs.table.headers.host"), + render: (row) => { + const host = row.workerId.split(":")[0]; + return ( + + {host} + + ); + }, + }, + { + id: "processId", + content: t("pages.logs.table.headers.processId"), + render: (row) => { + const processId = row.workerId.split(":")[1]; + return ( + + {processId} + + ); + }, + }, + { + id: "id", + content: t("pages.logs.table.headers.workerId"), + render: (row) => { + const workerId = row.workerId.split(":")[2]; + const shortWorkerId = workerId.slice(workerId.length - 8); + return ( + + {`...${shortWorkerId}`} + + ); + }, + }, + { + id: "mode", + content: t("pages.logs.table.headers.workerMode"), + render: (row) => {row.mode}, + }, + { + id: "lastSeen", + content: t("pages.logs.table.headers.lastSeenAt"), + render: (row) => ( + {formatTimestamp(row.lastSeenAt)} + ), + }, + ]; + + return ( + + ); +}; diff --git a/client/src/Pages/Logs/index.tsx b/client/src/Pages/Logs/index.tsx index b21e331c5..9d71e3253 100644 --- a/client/src/Pages/Logs/index.tsx +++ b/client/src/Pages/Logs/index.tsx @@ -8,7 +8,7 @@ import { useTranslation } from "react-i18next"; const LogsPage = () => { const { t } = useTranslation(); - const [activeTab, setActiveTab] = useState(0); + const [activeTab, setActiveTab] = useState(1); return ( Highlighted rows are jobs currently being processed by a worker.", "failedJobs": "Failed jobs", @@ -1000,6 +1002,11 @@ }, "table": { "headers": { + "host": "Host", + "processId": "Process ID", + "workerId": "Worker ID", + "workerMode": "Worker mode", + "lastSeenAt": "Last seen at", "state": "State", "timestamp": "Timestamp", "level": "Level", @@ -1010,6 +1017,7 @@ "failCount": "Count", "lastRunAt": "Last run at", "lockedAt": "Locked at", + "lockedBy": "Locked by", "lastFinishedAt": "Last finished at", "lastRunTook": "Last run took", "lastFailedAt": "Last failed", From b9fd29b906b70272ff2413c15ee62320af27030a Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 11:11:24 -0700 Subject: [PATCH 20/39] add compliant worker ID to super simple queue --- .../infrastructure/JobQueues/SuperSimpleQueue.ts | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts index 353a7ec7c..c96448728 100644 --- a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts @@ -3,6 +3,9 @@ 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"; + const SERVICE_NAME = "JobQueue"; type QueueJobFailure = { @@ -61,6 +64,7 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { private logger: ILogger; private helper: IQueueHelper; + private workerId: string; private monitorsRepository: IMonitorsRepository; private readonly scheduler: Scheduler; @@ -70,6 +74,7 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { this.monitorsRepository = monitorsRepository; this.scheduler = scheduler; this.registerListeners(); + this.workerId = `${hostname()}:${process.pid}:${randomUUID()}`; } get serviceName() { @@ -354,7 +359,13 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { }>, totalRuns: 0, totalFailures: 0, - workers: [], + workers: [ + { + workerId: this.workerId, + mode: "primary", + lastSeenAt: Date.now(), + }, + ], } ); return metrics; @@ -372,7 +383,7 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { monitorActive: job.data?.isActive ?? null, active: job.active, repeat: job.repeat ?? null, - lockedBy: null, + lockedBy: job.lockedAt ? this.workerId : null, lockedUntil: null, lockedAt: job.lockedAt ?? null, runCount: job.runCount ?? 0, From 2b6734bb783446ccf552717a2cd0af4737e682f6 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 11:25:21 -0700 Subject: [PATCH 21/39] update tests --- .../unit/services/superSimpleQueue.test.ts | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/server/test/unit/services/superSimpleQueue.test.ts b/server/test/unit/services/superSimpleQueue.test.ts index 970706199..3c2506774 100644 --- a/server/test/unit/services/superSimpleQueue.test.ts +++ b/server/test/unit/services/superSimpleQueue.test.ts @@ -486,10 +486,25 @@ describe("SuperSimpleQueue", () => { // ── getMetrics ─────────────────────────────────────────────────────────── describe("getMetrics", () => { - it("returns empty metrics when no jobs", async () => { + it("returns empty job metrics plus this process as the sole worker 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, workers: [] }); + expect(metrics).toEqual({ + jobs: 0, + activeJobs: 0, + failingJobs: 0, + jobsWithFailures: [], + totalRuns: 0, + totalFailures: 0, + workers: [ + { + // hostname:pid:uuid, matching less-simple-scheduler's workerId format + workerId: expect.stringMatching(/^.+:\d+:[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/), + mode: "primary", + lastSeenAt: expect.any(Number), + }, + ], + }); }); it("aggregates metrics from jobs", async () => { From dfae6c8abdf1669a66f8bc3105c6673dd4d9b2ac Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 11:26:10 -0700 Subject: [PATCH 22/39] type metrics for super simple queue --- .../JobQueues/SuperSimpleQueue.ts | 22 ++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts index c96448728..ee542d2be 100644 --- a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts @@ -5,6 +5,7 @@ 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"; const SERVICE_NAME = "JobQueue"; @@ -316,7 +317,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; From f5a4a038983080fd2fc33ab77b3ecb2a4cfb7e38 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 11:31:11 -0700 Subject: [PATCH 23/39] reorder stop --- .../infrastructure/JobQueues/LessSimpleQueue.ts | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts index cedf29901..ae0b5df72 100644 --- a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts @@ -27,6 +27,7 @@ export class LessSimpleQueue implements IJobQueue { private monitorsRepository: IMonitorsRepository; private workersRepository: IQueueWorkersRepository; private readonly scheduler: Scheduler; + private queueMode: QueueMode; private heartbeatListener: ((workerId: string) => void) | null = null; constructor( @@ -34,13 +35,15 @@ export class LessSimpleQueue implements IJobQueue { helper: IQueueHelper, monitorsRepository: IMonitorsRepository, workersRepository: IQueueWorkersRepository, - scheduler: Scheduler + scheduler: Scheduler, + queueMode: QueueMode ) { this.logger = logger; this.helper = helper; this.monitorsRepository = monitorsRepository; this.workersRepository = workersRepository; this.scheduler = scheduler; + this.queueMode = queueMode; this.registerListeners(); } @@ -137,7 +140,7 @@ export class LessSimpleQueue implements IJobQueue { const scheduler = new Scheduler(store, { concurrency: 50, }); - const instance = new LessSimpleQueue(logger, helper, monitorsRepository, workersRepository, scheduler); + const instance = new LessSimpleQueue(logger, helper, monitorsRepository, workersRepository, scheduler, queueMode); await instance.init(queueMode); return instance; @@ -422,8 +425,14 @@ export class LessSimpleQueue implements IJobQueue { }; 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), From b4970d321d27b23565e1e92019258d9127e12939 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Tue, 2 Jun 2026 11:31:22 -0700 Subject: [PATCH 24/39] update tests --- .../unit/services/lessSimpleQueue.test.ts | 28 +++++++++++++++++-- 1 file changed, 26 insertions(+), 2 deletions(-) diff --git a/server/test/unit/services/lessSimpleQueue.test.ts b/server/test/unit/services/lessSimpleQueue.test.ts index 1b455ef1b..19b68dee1 100644 --- a/server/test/unit/services/lessSimpleQueue.test.ts +++ b/server/test/unit/services/lessSimpleQueue.test.ts @@ -67,14 +67,15 @@ const createQueue = (overrides?: Record) => { const workersRepository = createWorkersRepo(); const scheduler = createScheduler(); - const defaults = { logger, helper, monitorsRepository, workersRepository, scheduler, ...overrides }; + const defaults = { logger, helper, monitorsRepository, workersRepository, scheduler, queueMode: "primary", ...overrides }; const queue = new LessSimpleQueue( defaults.logger as any, defaults.helper as any, defaults.monitorsRepository as any, defaults.workersRepository as any, - defaults.scheduler as any + defaults.scheduler as any, + defaults.queueMode as any ); return { queue, ...defaults }; }; @@ -686,6 +687,29 @@ describe("LessSimpleQueue", () => { expect(result.success).toBe(true); }); + it("flushes before stopping, so the store is still connected for removeAll", async () => { + const { queue, scheduler } = createQueue(); + const order: string[] = []; + (scheduler.flushJobs as jest.Mock).mockImplementation(async () => { + order.push("flush"); + return true; + }); + (scheduler.stop as jest.Mock).mockImplementation(async () => { + order.push("stop"); + return true; + }); + await queue.flushQueues(); + expect(order).toEqual(["flush", "stop"]); + }); + + it("refuses to flush on a non-primary worker without touching the scheduler", async () => { + const { queue, scheduler } = createQueue({ queueMode: "worker" }); + const result = await queue.flushQueues(); + expect(result.success).toBe(false); + expect(scheduler.flushJobs).not.toHaveBeenCalled(); + expect(scheduler.stop).not.toHaveBeenCalled(); + }); + it("returns false when flush fails", async () => { const { queue, scheduler } = createQueue(); (scheduler.flushJobs as jest.Mock).mockResolvedValue(false); From 01c3870b22d22a9d208d942cab8e3103ba43218f Mon Sep 17 00:00:00 2001 From: Br0wnHammer Date: Wed, 3 Jun 2026 20:15:01 +0530 Subject: [PATCH 25/39] feat(client): enable step wizard on infrastructure and pagespeed create pages --- client/src/Pages/CreateMonitor/index.tsx | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/client/src/Pages/CreateMonitor/index.tsx b/client/src/Pages/CreateMonitor/index.tsx index f34876359..98510b57c 100644 --- a/client/src/Pages/CreateMonitor/index.tsx +++ b/client/src/Pages/CreateMonitor/index.tsx @@ -234,6 +234,9 @@ const CreateMonitorPage = () => { }, [location.pathname]); const showTypeSelector = pageType === "uptime" && !isEditMode; + // The multi-step wizard applies to every create flow; only the monitor-type + // radio is uptime-specific. Edit mode keeps the flat form. + const isWizard = !isEditMode; const defaultType: MonitorType = pageType === "pagespeed" ? "pagespeed" @@ -267,7 +270,7 @@ const CreateMonitorPage = () => { // Multi-step wizard, uptime create only (showTypeSelector). Other flows // keep the flat form, where showStep() is always true. const [currentStep, setCurrentStep] = useState(0); - const showStep = (step: number) => !showTypeSelector || currentStep === step; + const showStep = (step: number) => !isWizard || currentStep === step; const watchedType = watch("type") as MonitorType; const watchedUseAdvancedMatching = watch("useAdvancedMatching") as boolean; @@ -362,7 +365,7 @@ const CreateMonitorPage = () => { refetch={refetchMonitor} onDelete={handleDeleteClick} /> - {showTypeSelector && ( + {isWizard && ( { - {showTypeSelector && ( + {isWizard && ( )} - {showTypeSelector && currentStep < totalSteps - 1 ? ( + {isWizard && currentStep < totalSteps - 1 ? (
({ - ...(row.lockedAt && { - "& td": { backgroundColor: theme.palette.rowStatus.running }, - }), - ...(row.monitorActive === false && { - "& td": { backgroundColor: theme.palette.rowStatus.paused }, - }), - })} - /> + <> +
({ + ...(row.lockedAt && { + "& td": { backgroundColor: theme.palette.rowStatus.running }, + }), + ...(row.monitorActive === false && { + "& td": { backgroundColor: theme.palette.rowStatus.paused }, + }), + })} + /> + + ); }; diff --git a/client/src/Types/Queue.ts b/client/src/Types/Queue.ts index d7c1d15e2..5a9ae7cdf 100644 --- a/client/src/Types/Queue.ts +++ b/client/src/Types/Queue.ts @@ -51,5 +51,6 @@ export type QueueJobSummary = { export interface QueueData { jobs: QueueJobSummary[]; + count: number; metrics: QueueMetrics; } From 3c85b36b1bd9e8efc7c9777ea8f4d30e50888460 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 3 Jun 2026 13:30:03 -0700 Subject: [PATCH 38/39] add pagnation validation to queue query --- server/src/validation/queueValidation.ts | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 server/src/validation/queueValidation.ts diff --git a/server/src/validation/queueValidation.ts b/server/src/validation/queueValidation.ts new file mode 100644 index 000000000..61c552751 --- /dev/null +++ b/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(), +}); From 7872b18108d534b8ad27b700e475feaff3dffdd8 Mon Sep 17 00:00:00 2001 From: Br0wnHammer Date: Thu, 4 Jun 2026 11:09:09 +0530 Subject: [PATCH 39/39] refactor(client): inline isEditMode instead of shadowing it with isWizard --- client/src/Pages/CreateMonitor/index.tsx | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/client/src/Pages/CreateMonitor/index.tsx b/client/src/Pages/CreateMonitor/index.tsx index 98510b57c..94bed9355 100644 --- a/client/src/Pages/CreateMonitor/index.tsx +++ b/client/src/Pages/CreateMonitor/index.tsx @@ -234,9 +234,6 @@ const CreateMonitorPage = () => { }, [location.pathname]); const showTypeSelector = pageType === "uptime" && !isEditMode; - // The multi-step wizard applies to every create flow; only the monitor-type - // radio is uptime-specific. Edit mode keeps the flat form. - const isWizard = !isEditMode; const defaultType: MonitorType = pageType === "pagespeed" ? "pagespeed" @@ -267,10 +264,10 @@ const CreateMonitorPage = () => { reset(defaults); }, [defaults, reset]); - // Multi-step wizard, uptime create only (showTypeSelector). Other flows - // keep the flat form, where showStep() is always true. + // Multi-step wizard on every create flow; edit mode keeps the flat form + // where showStep() is always true. const [currentStep, setCurrentStep] = useState(0); - const showStep = (step: number) => !isWizard || currentStep === step; + const showStep = (step: number) => isEditMode || currentStep === step; const watchedType = watch("type") as MonitorType; const watchedUseAdvancedMatching = watch("useAdvancedMatching") as boolean; @@ -365,7 +362,7 @@ const CreateMonitorPage = () => { refetch={refetchMonitor} onDelete={handleDeleteClick} /> - {isWizard && ( + {!isEditMode && ( { - {isWizard && ( + {!isEditMode && ( )} - {isWizard && currentStep < totalSteps - 1 ? ( + {!isEditMode && currentStep < totalSteps - 1 ? (