From c3944f2fe701fd8e4669036b669b771946823390 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 3 Jun 2026 13:28:15 -0700 Subject: [PATCH 1/6] Add pagination to controller --- server/src/controllers/queueController.ts | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/server/src/controllers/queueController.ts b/server/src/controllers/queueController.ts index 52762f42b..3b0bb64ce 100755 --- a/server/src/controllers/queueController.ts +++ b/server/src/controllers/queueController.ts @@ -1,4 +1,5 @@ -import { ISuperSimpleQueue } from "@/service/index.js"; +import { IJobQueue } from "@/service/index.js"; +import { getQueueJobsQueryValidation } from "@/validation/queueValidation.js"; import { Request, Response, NextFunction } from "express"; const SERVICE_NAME = "JobQueueController"; @@ -12,8 +13,8 @@ export interface IJobQueueController { class JobQueueController implements IJobQueueController { static SERVICE_NAME = SERVICE_NAME; - private jobQueue: ISuperSimpleQueue; - constructor(jobQueue: ISuperSimpleQueue) { + private jobQueue: IJobQueue; + constructor(jobQueue: IJobQueue) { this.jobQueue = jobQueue; } @@ -36,11 +37,12 @@ class JobQueueController implements IJobQueueController { getJobs = async (req: Request, res: Response, next: NextFunction) => { try { - const jobs = await this.jobQueue.getJobs(); + const pagination = getQueueJobsQueryValidation.parse(req.query); + const { jobs, count } = await this.jobQueue.getJobs(pagination); return res.status(200).json({ success: true, msg: "Queue jobs fetched successfully", - data: jobs, + data: { jobs, count }, }); } catch (error) { next(error); @@ -49,12 +51,13 @@ class JobQueueController implements IJobQueueController { getAllMetrics = async (req: Request, res: Response, next: NextFunction) => { try { - const jobs = await this.jobQueue.getJobs(); + const pagination = getQueueJobsQueryValidation.parse(req.query); + const { jobs, count } = await this.jobQueue.getJobs(pagination); const metrics = await this.jobQueue.getMetrics(); return res.status(200).json({ success: true, msg: "Queue metrics fetched successfully", - data: { jobs, metrics }, + data: { jobs, count, metrics }, }); } catch (error) { next(error); From 63fd3f6afd748c9b879f8da9a3af2a042637f181 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 3 Jun 2026 13:28:25 -0700 Subject: [PATCH 2/6] bump 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 72a485391..0f16e6b5d 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.4.1", + "less-simple-scheduler": "2.6.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.4.1", - "resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.4.1.tgz", - "integrity": "sha512-7Bgd5qLahFT3rv2QlY8QHrkkXLfUUgUpdDyabAiiy0WeEn1ay7Sacxcbge9TCZ34ZsySp+h1+NHeoTnWVU0vaQ==", + "version": "2.6.0", + "resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.6.0.tgz", + "integrity": "sha512-Vi8ouiodLqe2fyh+DqCmxE5cCJZDI3xOyUAH4jcj9sf7GunLUzH71PBcYJy2P8lxTK3LxEkHtr1reymiogs6JQ==", "license": "ISC", "dependencies": { "mongodb": "7.2.0" diff --git a/server/package.json b/server/package.json index c4ed47933..2812b9b1f 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.4.1", + "less-simple-scheduler": "2.6.0", "mailersend": "^2.2.0", "mjml": "^5.0.0-alpha.4", "mongoose": "^8.3.3", From 174fb048f739bb449253eff3737adf93214d02a9 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 3 Jun 2026 13:28:53 -0700 Subject: [PATCH 3/6] implement pagination, update metrics to use scheduler's aggregation --- .../infrastructure/JobQueues/IJobQueue.ts | 12 ++- .../JobQueues/LessSimpleQueue.ts | 95 +++++++------------ .../JobQueues/SuperSimpleQueue.ts | 64 ++----------- 3 files changed, 57 insertions(+), 114 deletions(-) diff --git a/server/src/service/infrastructure/JobQueues/IJobQueue.ts b/server/src/service/infrastructure/JobQueues/IJobQueue.ts index 371a7c74a..ccbca625a 100644 --- a/server/src/service/infrastructure/JobQueues/IJobQueue.ts +++ b/server/src/service/infrastructure/JobQueues/IJobQueue.ts @@ -41,6 +41,16 @@ export type QueueJobSummary = { repeat: number | null; }; +export type QueueJobsPagination = { + page?: number; + rowsPerPage?: number; +}; + +export type QueueJobsPage = { + jobs: QueueJobSummary[]; + count: number; +}; + export interface IJobQueue { readonly serviceName: string; init(): Promise; @@ -51,6 +61,6 @@ export interface IJobQueue { updateJob(monitor: Monitor): Promise; shutdown(): Promise; getMetrics(): Promise; - getJobs(): Promise; + getJobs(pagination: QueueJobsPagination): Promise; flushQueues(): Promise<{ success: boolean }>; } diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts index 51dcd9460..963b41267 100644 --- a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts @@ -1,22 +1,16 @@ import { IMonitorsRepository, IQueueWorkersRepository } from "@/repositories/index.js"; import { ILogger } from "@/utils/logger.js"; -import { MongoStore, Scheduler, IJob } from "less-simple-scheduler"; +import { MongoStore, Scheduler, IJob, AddJobInput } from "less-simple-scheduler"; import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js"; import { Monitor, supportsGeoCheck } from "@/types/monitor.js"; -import { type QueueMetrics, IJobQueue } from "@/service/infrastructure/JobQueues/IJobQueue.js"; +import { type QueueMetrics, type QueueJobsPagination, type QueueJobsPage, IJobQueue } from "@/service/infrastructure/JobQueues/IJobQueue.js"; import { type QueueMode } from "@/types/settings.js"; import { EnvConfig } from "@/service/system/settingsService.js"; const SERVICE_NAME = "JobQueue"; -// How long after its last heartbeat a worker is still considered alive. 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. +// How long after its last heartbeat a worker is still considered alive. Must be longer than the scheduler's heartbeat const WORKER_TTL_MS = 30_000; -// Derived from the scheduler -type SchedulerJobInput = Parameters[0][number]; - // For discriminating job types type MonitorJob = Omit & { data: Monitor }; const MONITOR_TEMPLATES = new Set(["monitor-job", "geo-check-job"]); @@ -215,8 +209,8 @@ export class LessSimpleQueue implements IJobQueue { // Builds the scheduler inputs for a monitor: always a monitor-job, plus a // geo-check job when the monitor supports and enables geo checks. - private buildJobInputs = (monitor: Monitor): SchedulerJobInput[] => { - const inputs: SchedulerJobInput[] = [ + private buildJobInputs = (monitor: Monitor): AddJobInput[] => { + const inputs: AddJobInput[] = [ { id: monitor.id, template: "monitor-job", @@ -355,58 +349,39 @@ export class LessSimpleQueue implements IJobQueue { this.scheduler.stop(); }; - getMetrics = async () => { - const jobs = await this.scheduler.getJobs(); - const now = Date.now(); + getMetrics = async (): Promise => { + const stats = await this.scheduler.getStats(); - const metrics = jobs.reduce( - (acc, job) => { - const runCount = job.runCount ?? 0; - const failCount = job.failCount ?? 0; - const lastFailedAt = job.lastFailedAt ?? 0; - const lastFinishedAt = job.lastFinishedAt ?? 0; - acc.totalRuns += runCount; - acc.totalFailures += failCount; - acc.jobs++; - if (failCount > 0 && lastFailedAt >= lastFinishedAt) { - acc.failingJobs++; - } + const jobsWithFailures = stats.jobsWithFailures.map((failure) => { + // `data` is untyped at the store level: for monitor jobs it's the Monitor, + // for system jobs (e.g. cleanup) it has no url/type => null + const monitor = failure.data as Monitor | undefined; + return { + monitorId: failure.id, + monitorUrl: monitor?.url ?? null, + monitorType: monitor?.type ?? null, + failedAt: failure.failedAt, + failCount: failure.failCount, + failReason: failure.failReason, + }; + }); - // A job currently held by a live lock is in flight on some worker - if (job.lockedBy && job.lockedUntil && job.lockedUntil > now) { - acc.activeJobs++; - } - - if (failCount > 0) { - const monitor = isMonitorJob(job) ? job.data : undefined; - acc.jobsWithFailures.push({ - monitorId: job.id, - monitorUrl: monitor?.url || null, - monitorType: monitor?.type || null, - failedAt: job.lastFailedAt ?? null, - failCount, - failReason: job.lastError ?? null, - }); - } - return acc; - }, - { - jobs: 0, - activeJobs: 0, - failingJobs: 0, - jobsWithFailures: [], - totalRuns: 0, - totalFailures: 0, - workers: [], - } - ); - metrics.workers = await this.workersRepository.findRecent(WORKER_TTL_MS); - return metrics; + const workers = await this.workersRepository.findRecent(WORKER_TTL_MS); + return { + jobs: stats.jobs, + activeJobs: stats.activeJobs, + failingJobs: stats.failingJobs, + totalRuns: stats.totalRuns, + totalFailures: stats.totalFailures, + jobsWithFailures, + workers, + }; }; - getJobs = async () => { - const jobs = await this.scheduler.getJobs(); - return jobs.map((job) => { + getJobs = async ({ page = 0, rowsPerPage = 0 }: QueueJobsPagination): Promise => { + const listOptions = rowsPerPage > 0 ? { skip: Math.max(page, 0) * rowsPerPage, limit: rowsPerPage } : undefined; + const [jobs, count] = await Promise.all([this.scheduler.getJobs(listOptions), this.scheduler.countJobs()]); + const mapped = jobs.map((job) => { const monitor = isMonitorJob(job) ? job.data : undefined; return { @@ -430,6 +405,8 @@ export class LessSimpleQueue implements IJobQueue { lastFailedAt: job.lastFailedAt ?? null, }; }); + + return { jobs: mapped, count }; }; flushQueues = async () => { diff --git a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts index ee542d2be..42409696b 100644 --- a/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts @@ -6,61 +6,11 @@ import { Monitor, MonitorType, supportsGeoCheck } from "@/types/monitor.js"; import { hostname } from "node:os"; import { randomUUID } from "node:crypto"; import { QueueMode } from "@/types/settings.js"; +import { IJobQueue, type QueueJobSummary } from "@/service/infrastructure/JobQueues/IJobQueue.js"; const SERVICE_NAME = "JobQueue"; -type QueueJobFailure = { - monitorId: string | number; - monitorUrl: string | null; - monitorType: string | null; - failedAt: number | null; - failCount: number; - failReason: string | null; -}; - -type QueueMetrics = { - jobs: number; - activeJobs: number; - failingJobs: number; - jobsWithFailures: QueueJobFailure[]; - totalRuns: number; - totalFailures: number; -}; - -type QueueJobSummary = { - monitorId: string | number; - monitorUrl: string | null; - monitorType: string | null; - monitorInterval: number | null; - monitorGeoInterval: number | null; - monitorActive: boolean | null; - active: boolean; - lockedAt: number | null; - runCount: number; - failCount: number; - failReason: string | null; - lastRunAt: number | null; - lastFinishedAt: number | null; - lastRunTook: number | null; - lastFailedAt: number | null; - repeat: number | null; -}; - -export interface ISuperSimpleQueue { - readonly serviceName: string; - init(): Promise; - addJob(monitorId: string, monitor: Monitor): Promise; - deleteJob(monitor: Monitor): Promise; - pauseJob(monitor: Monitor): Promise; - resumeJob(monitor: Monitor): Promise; - updateJob(monitor: Monitor): Promise; - shutdown(): Promise; - getMetrics(): Promise; - getJobs(): Promise; - flushQueues(): Promise<{ success: boolean }>; -} - -export class SuperSimpleQueue implements ISuperSimpleQueue { +export class SuperSimpleQueue implements IJobQueue { static SERVICE_NAME = SERVICE_NAME; private logger: ILogger; @@ -391,9 +341,9 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { return metrics; }; - getJobs = async () => { + getJobs = async ({ page = 0, rowsPerPage = 0 }: { page?: number; rowsPerPage?: number }): Promise<{ jobs: QueueJobSummary[]; count: number }> => { const jobs = await this.scheduler.getJobs(); - return jobs.map((job) => { + const mapped = jobs.map((job) => { return { monitorId: job.id, monitorUrl: job.data?.url || null, @@ -415,6 +365,12 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { lastFailedAt: job.lastFailedAt ?? null, }; }); + + const start = Math.max(page, 0) * rowsPerPage; + return { + jobs: rowsPerPage > 0 ? mapped.slice(start, start + rowsPerPage) : mapped, + count: mapped.length, + }; }; flushQueues = async () => { From 47298638260bd938605013f88ddc52929134574b Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 3 Jun 2026 13:29:03 -0700 Subject: [PATCH 4/6] add tests --- .../unit/services/lessSimpleQueue.test.ts | 165 ++++++++---------- .../unit/services/superSimpleQueue.test.ts | 33 +++- 2 files changed, 96 insertions(+), 102 deletions(-) diff --git a/server/test/unit/services/lessSimpleQueue.test.ts b/server/test/unit/services/lessSimpleQueue.test.ts index 0b1ec5bbc..6351dc5d5 100644 --- a/server/test/unit/services/lessSimpleQueue.test.ts +++ b/server/test/unit/services/lessSimpleQueue.test.ts @@ -26,6 +26,8 @@ const createScheduler = () => { removeJob: jest.fn(), getJob: jest.fn().mockResolvedValue(null), getJobs: jest.fn().mockResolvedValue([]), + countJobs: jest.fn().mockResolvedValue(0), + getStats: jest.fn().mockResolvedValue({ jobs: 0, activeJobs: 0, failingJobs: 0, totalRuns: 0, totalFailures: 0, jobsWithFailures: [] }), pauseJob: jest.fn().mockResolvedValue(true), resumeJob: jest.fn().mockResolvedValue(true), updateJob: jest.fn(), @@ -81,9 +83,6 @@ const createQueue = (overrides?: Record) => { 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", @@ -484,107 +483,55 @@ describe("LessSimpleQueue", () => { expect(metrics).toEqual({ jobs: 0, activeJobs: 0, failingJobs: 0, jobsWithFailures: [], totalRuns: 0, totalFailures: 0, workers: [] }); }); - it("aggregates metrics from jobs", async () => { + it("maps the store's aggregate counters through", 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, - // live lock held by a worker - lockedBy: "host-1:1234:abcd", - lockedUntil: FAR_FUTURE, - lockedAt: 999, - data: { url: "http://b.com", type: "ping" }, - }, - ]); + (scheduler.getStats as jest.Mock).mockResolvedValue({ + jobs: 2, + activeJobs: 1, + failingJobs: 1, + totalRuns: 15, + totalFailures: 2, + jobsWithFailures: [], + }); 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" }) - ); + expect(metrics).toMatchObject({ jobs: 2, activeJobs: 1, failingJobs: 1, totalRuns: 15, totalFailures: 2 }); }); - it("does not count as failing when lastFailedAt < lastFinishedAt", async () => { + it("derives monitor fields for failing monitor jobs from job data", 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" } }, - ]); + (scheduler.getStats as jest.Mock).mockResolvedValue({ + jobs: 1, + activeJobs: 0, + failingJobs: 1, + totalRuns: 10, + totalFailures: 2, + jobsWithFailures: [{ id: "m1", data: { url: "http://a.com", type: "http" }, failedAt: 200, failCount: 2, failReason: "timeout" }], + }); const metrics = await queue.getMetrics(); - expect(metrics.failingJobs).toBe(0); - expect(metrics.jobsWithFailures).toHaveLength(1); + expect(metrics.jobsWithFailures).toEqual([ + { monitorId: "m1", monitorUrl: "http://a.com", monitorType: "http", failedAt: 200, failCount: 2, failReason: "timeout" }, + ]); }); - it("leaves monitor fields null for non-monitor (cleanup) jobs with failures", async () => { + it("leaves monitor fields null for non-monitor (cleanup) 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, - }, - ]); + (scheduler.getStats as jest.Mock).mockResolvedValue({ + jobs: 1, + activeJobs: 0, + failingJobs: 1, + totalRuns: 1, + totalFailures: 1, + jobsWithFailures: [{ id: "cleanup-orphaned", data: undefined, failedAt: 100, failCount: 1, failReason: "boom" }], + }); const metrics = await queue.getMetrics(); - expect(metrics.jobsWithFailures[0].monitorUrl).toBeNull(); - expect(metrics.jobsWithFailures[0].monitorType).toBeNull(); - expect(metrics.jobsWithFailures[0].failReason).toBe("boom"); - }); - }); - - // ── 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); + expect(metrics.jobsWithFailures[0]).toEqual({ + monitorId: "cleanup-orphaned", + monitorUrl: null, + monitorType: null, + failedAt: 100, + failCount: 1, + failReason: "boom", + }); }); }); @@ -627,7 +574,7 @@ describe("LessSimpleQueue", () => { data: { url: "http://a.com", type: "http", interval: 60000, geoCheckInterval: 300000, isActive: true }, }, ]); - const jobs = await queue.getJobs(); + const { jobs } = await queue.getJobs({}); expect(jobs).toHaveLength(1); expect(jobs[0]).toEqual({ @@ -666,20 +613,44 @@ describe("LessSimpleQueue", () => { data: { url: "a", type: "http", interval: 60000 }, }, ]); - const jobs = await queue.getJobs(); + 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(); + 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(); }); + + it("pushes pagination to the scheduler and returns its total count", async () => { + const { queue, scheduler } = createQueue(); + (scheduler.getJobs as jest.Mock).mockResolvedValue([ + { id: "m1", template: "monitor-job", active: true, data: { url: "a", type: "http", interval: 60000 } }, + ]); + (scheduler.countJobs as jest.Mock).mockResolvedValue(42); + + const { jobs, count } = await queue.getJobs({ page: 2, rowsPerPage: 10 }); + + expect(scheduler.getJobs).toHaveBeenCalledWith({ skip: 20, limit: 10 }); + expect(count).toBe(42); + expect(jobs).toHaveLength(1); + }); + + it("requests all jobs when rowsPerPage is omitted", async () => { + const { queue, scheduler } = createQueue(); + (scheduler.countJobs as jest.Mock).mockResolvedValue(7); + + const { count } = await queue.getJobs({}); + + expect(scheduler.getJobs).toHaveBeenCalledWith(undefined); + expect(count).toBe(7); + }); }); // ── flushQueues ──────────────────────────────────────────────────────────── diff --git a/server/test/unit/services/superSimpleQueue.test.ts b/server/test/unit/services/superSimpleQueue.test.ts index 3c2506774..b6da507ce 100644 --- a/server/test/unit/services/superSimpleQueue.test.ts +++ b/server/test/unit/services/superSimpleQueue.test.ts @@ -583,7 +583,7 @@ describe("SuperSimpleQueue", () => { data: { url: "http://a.com", type: "http", interval: 60000, geoCheckInterval: 300000, isActive: true }, }, ]); - const jobs = await queue.getJobs(); + const { jobs } = await queue.getJobs({}); expect(jobs).toHaveLength(1); expect(jobs[0]).toEqual({ @@ -613,14 +613,14 @@ describe("SuperSimpleQueue", () => { (scheduler.getJobs as jest.Mock).mockResolvedValue([ { id: "m1", active: true, lockedAt: 999, runCount: 1, data: { url: "a", type: "http", interval: 60000 } }, ]); - const jobs = await queue.getJobs(); + const { jobs } = await queue.getJobs({}); expect(jobs[0].lastRunTook).toBeNull(); }); it("handles jobs with missing data", async () => { const { queue, scheduler } = createQueue(); (scheduler.getJobs as jest.Mock).mockResolvedValue([{ id: "m1", active: true, data: null }]); - const jobs = await queue.getJobs(); + const { jobs } = await queue.getJobs({}); expect(jobs[0].monitorUrl).toBeNull(); expect(jobs[0].monitorType).toBeNull(); expect(jobs[0].monitorInterval).toBeNull(); @@ -635,7 +635,7 @@ describe("SuperSimpleQueue", () => { { id: "m1", active: true, data: { url: "a", type: "http", interval: 60000, isActive: true } }, { id: "m2", active: false, data: { url: "b", type: "http", interval: 60000, isActive: false } }, ]); - const jobs = await queue.getJobs(); + const { jobs } = await queue.getJobs({}); expect(jobs[0].monitorActive).toBe(true); expect(jobs[1].monitorActive).toBe(false); }); @@ -645,10 +645,33 @@ describe("SuperSimpleQueue", () => { (scheduler.getJobs as jest.Mock).mockResolvedValue([ { id: "m1", active: true, repeat: 120000, data: { url: "a", type: "http", interval: 60000, geoCheckInterval: 600000 } }, ]); - const jobs = await queue.getJobs(); + const { jobs } = await queue.getJobs({}); expect(jobs[0].repeat).toBe(120000); expect(jobs[0].monitorGeoInterval).toBe(600000); }); + + it("paginates jobs and returns the total count", async () => { + const { queue, scheduler } = createQueue(); + (scheduler.getJobs as jest.Mock).mockResolvedValue( + Array.from({ length: 25 }, (_, i) => ({ id: `m${i}`, active: true, data: { url: `a${i}`, type: "http", interval: 60000 } })) + ); + + const { jobs, count } = await queue.getJobs({ page: 1, rowsPerPage: 10 }); + expect(count).toBe(25); + expect(jobs).toHaveLength(10); + expect(jobs[0].monitorId).toBe("m10"); + }); + + it("returns all jobs when rowsPerPage is omitted", async () => { + const { queue, scheduler } = createQueue(); + (scheduler.getJobs as jest.Mock).mockResolvedValue( + Array.from({ length: 25 }, (_, i) => ({ id: `m${i}`, active: true, data: { url: `a${i}`, type: "http", interval: 60000 } })) + ); + + const { jobs, count } = await queue.getJobs({}); + expect(count).toBe(25); + expect(jobs).toHaveLength(25); + }); }); // ── flushQueues ────────────────────────────────────────────────────────── From ee0bc5e500abdceed8e309be917c69459db3fc5b Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Wed, 3 Jun 2026 13:29:17 -0700 Subject: [PATCH 5/6] add pagination components --- client/src/Pages/Logs/TabQueue.tsx | 22 +++++++- .../src/Pages/Logs/components/TableJobs.tsx | 52 ++++++++++++++----- client/src/Types/Queue.ts | 1 + 3 files changed, 59 insertions(+), 16 deletions(-) diff --git a/client/src/Pages/Logs/TabQueue.tsx b/client/src/Pages/Logs/TabQueue.tsx index bb3aac368..761b7def8 100644 --- a/client/src/Pages/Logs/TabQueue.tsx +++ b/client/src/Pages/Logs/TabQueue.tsx @@ -4,6 +4,7 @@ import Box from "@mui/material/Box"; import { TableJobs, TableFailedJobs } from "./components/TableJobs"; import { TableWorkers } from "./components/TableWorkers"; import { useTheme } from "@mui/material"; +import { useState } from "react"; import { useTranslation, Trans } from "react-i18next"; import { useGet, usePost } from "@/Hooks/UseApi"; import type { QueueData } from "@/Types/Queue"; @@ -15,16 +16,23 @@ import { SPACING } from "@/Utils/Theme/constants"; export const TabQueue = () => { const theme = useTheme(); const { t } = useTranslation(); + const [page, setPage] = useState(0); + const [rowsPerPage, setRowsPerPage] = useState(10); const { data: queueData, isLoading, // error, refetch, - } = useGet("/queue/all-metrics", {}, { refreshInterval: 1000 }); + } = useGet( + `/queue/all-metrics?page=${page}&rowsPerPage=${rowsPerPage}`, + {}, + { refreshInterval: 1000, keepPreviousData: true } + ); const { post, loading: isFlushing } = usePost(); const jobs = queueData?.jobs ?? []; + const jobsCount = queueData?.count ?? 0; const workers = (queueData?.metrics?.workers ?? []).map((worker, idx) => ({ ...worker, id: idx, @@ -104,7 +112,17 @@ export const TabQueue = () => { /> - + setPage(newPage)} + onRowsPerPageChange={(event) => { + setRowsPerPage(parseInt(event.target.value, 10)); + setPage(0); + }} + /> diff --git a/client/src/Pages/Logs/components/TableJobs.tsx b/client/src/Pages/Logs/components/TableJobs.tsx index 25e537823..241b6f016 100644 --- a/client/src/Pages/Logs/components/TableJobs.tsx +++ b/client/src/Pages/Logs/components/TableJobs.tsx @@ -1,4 +1,4 @@ -import { Table } from "@/Components/design-elements"; +import { Table, Pagination } from "@/Components/design-elements"; import { Typography, useTheme } from "@mui/material"; import prettyMilliseconds from "pretty-ms"; import { formatTimestamp } from "@/Utils/TimeUtils"; @@ -6,14 +6,27 @@ import { formatTimestamp } from "@/Utils/TimeUtils"; import { useTranslation } from "react-i18next"; import type { QueueJobFailure, QueueJobSummary, QueueMetrics } from "@/Types/Queue"; import type { Header } from "@/Components/design-elements"; +import type { TablePaginationProps } from "@mui/material/TablePagination"; type QueueJobWithId = QueueJobSummary & { id: string | number }; interface TableJobsProps { jobs: QueueJobSummary[]; + count: number; + page: number; + rowsPerPage: number; + onPageChange: TablePaginationProps["onPageChange"]; + onRowsPerPageChange: TablePaginationProps["onRowsPerPageChange"]; } -export const TableJobs = ({ jobs }: TableJobsProps) => { +export const TableJobs = ({ + jobs, + count, + page, + rowsPerPage, + onPageChange, + onRowsPerPageChange, +}: TableJobsProps) => { const { t } = useTranslation(); const theme = useTheme(); @@ -151,18 +164,29 @@ export const TableJobs = ({ jobs }: TableJobsProps) => { ]; return ( - ({ - ...(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 6/6] 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(), +});