udpate tests

server/test/integration/jobsRepository.test.ts

@@ -497,4 +497,40 @@ describe("MongoJobsRepository", () => {
 			expect(await repo.findAll()).toHaveLength(2);
 		});
 	});
+
+	// ── countDueBacklog (scaling signal) ─────────────────────────────────────────
+	describe("countDueBacklog", () => {
+		it("counts a due, active, unlocked check job", async () => {
+			await seedJob();
+			expect(await repo.countDueBacklog(NOW)).toBe(1);
+		});
+
+		it("counts a check job whose lock has expired (reclaimable work)", async () => {
+			await seedJob({ lockedBy: "dead-worker", lockedUntil: NOW - 1 });
+			expect(await repo.countDueBacklog(NOW)).toBe(1);
+		});
+
+		it("excludes not-yet-due, inactive, live-locked, and non-check rows", async () => {
+			await seedJob({ _id: "check:due", refId: "due" }); // counts
+			await seedJob({ _id: "check:future", refId: "future", nextScheduledAt: NOW + 1 }); // not due
+			await seedJob({ _id: "check:paused", refId: "paused", isActive: false }); // inactive
+			await seedJob({ _id: "check:locked", refId: "locked", lockedBy: "live", lockedUntil: NOW + LOCK_MS }); // live lock
+			await seedJob({ _id: "geo:mon", refId: "mon", type: "geo-check" }); // wrong type
+			await seedJob({ _id: "evaluate:mon", refId: "mon", type: "evaluate" }); // wrong type
+
+			expect(await repo.countDueBacklog(NOW)).toBe(1); // only check:due
+		});
+
+		it("counts every qualifying check row across monitors", async () => {
+			await seedJob({ _id: "check:a", refId: "a" });
+			await seedJob({ _id: "check:b", refId: "b" });
+			await seedJob({ _id: "check:c", refId: "c", lockedBy: "live", lockedUntil: NOW + LOCK_MS }); // excluded
+			expect(await repo.countDueBacklog(NOW)).toBe(2);
+		});
+
+		it("returns 0 when nothing is due", async () => {
+			await seedJob({ nextScheduledAt: NOW + 10_000 });
+			expect(await repo.countDueBacklog(NOW)).toBe(0);
+		});
+	});
 });

server/test/integration/schedulerReconcile.test.ts

@@ -0,0 +1,107 @@
+import { describe, expect, it, beforeAll, afterAll, beforeEach } from "@jest/globals";
+import mongoose from "mongoose";
+import { MongoMemoryServer } from "mongodb-memory-server";
+import { JobScheduler } from "../../src/worker/worker.job-scheduler.ts";
+import MongoJobsRepository from "../../src/domain/jobs/job.repository.mongo.ts";
+import MongoQueueWorkersRepository from "../../src/domain/queue-workers/queue-worker.repository.mongo.ts";
+import MongoMonitorsRepository from "../../src/domain/monitors/monitor.repository.mongo.ts";
+import JobModel, { type JobDocument } from "../../src/domain/jobs/job.model.ts";
+import { MonitorModel } from "../../src/domain/monitors/monitor.model.ts";
+import { QueueWorkerModel } from "../../src/domain/queue-workers/queue-worker.model.ts";
+import { createMockLogger } from "../helpers/createMockLogger.ts";
+
+// ── Real-Mongo harness ─────────────────────────────────────────────────────────
+// The bug: a scheduler-only primary (queueMode=primary, queuePrimaryProcesses=false)
+// never seeded the jobs collection, so a fresh start with existing monitors left the
+// workers idle. The fix moved reconcile/heartbeat onto the base JobScheduler.init().
+// This proves init() against a live mongod actually writes the expected job rows.
+
+let mongod: MongoMemoryServer;
+
+beforeAll(async () => {
+	mongod = await MongoMemoryServer.create();
+	await mongoose.connect(mongod.getUri());
+	await JobModel.init();
+}, 120_000);
+
+afterAll(async () => {
+	await mongoose.disconnect();
+	await mongod.stop();
+});
+
+beforeEach(async () => {
+	await JobModel.deleteMany({});
+	await MonitorModel.collection.deleteMany({});
+	await QueueWorkerModel.deleteMany({});
+});
+
+const WORKER_ID = "scheduler-primary-1";
+
+const seedMonitor = (overrides: Record<string, unknown> = {}) =>
+	MonitorModel.create({
+		userId: new mongoose.Types.ObjectId(),
+		teamId: new mongoose.Types.ObjectId(),
+		name: "mon",
+		type: "http",
+		url: "https://example.com",
+		interval: 60_000,
+		isActive: true,
+		...overrides,
+	});
+
+const makeScheduler = () =>
+	new JobScheduler(
+		new MongoJobsRepository(WORKER_ID),
+		new MongoQueueWorkersRepository(),
+		new MongoMonitorsRepository(),
+		"primary",
+		createMockLogger(),
+		WORKER_ID
+	);
+
+const idsOf = async () => (await JobModel.find().lean<JobDocument[]>()).map((j) => j._id);
+
+describe("JobScheduler.init (scheduler-only primary reconcile)", () => {
+	it("seeds a check row per monitor plus the two cleanup rows", async () => {
+		const m1 = await seedMonitor();
+		const m2 = await seedMonitor({ name: "mon-2" });
+
+		const scheduler = makeScheduler();
+		await scheduler.init();
+		await scheduler.shutdown(); // stop the heartbeat interval
+
+		const ids = await idsOf();
+		expect(ids).toContain(`check:${m1.id}`);
+		expect(ids).toContain(`check:${m2.id}`);
+		expect(ids).toContain("cleanup-orphaned");
+		expect(ids).toContain("cleanup-retention");
+	});
+
+	it("seeds a geo-check row only for geo-enabled monitors", async () => {
+		const geo = await seedMonitor({ name: "geo", geoCheckEnabled: true });
+		const plain = await seedMonitor({ name: "plain", geoCheckEnabled: false });
+
+		const scheduler = makeScheduler();
+		await scheduler.init();
+		await scheduler.shutdown();
+
+		const ids = await idsOf();
+		expect(ids).toContain(`geo-check:${geo.id}`);
+		expect(ids).not.toContain(`geo-check:${plain.id}`);
+	});
+
+	it("registers the primary in the queue_workers registry", async () => {
+		await seedMonitor();
+
+		const scheduler = makeScheduler();
+		await scheduler.init();
+		await scheduler.shutdown();
+
+		// shutdown deregisters, so assert mid-flight: re-init then read before shutdown.
+		const scheduler2 = makeScheduler();
+		await scheduler2.init();
+		const row = await QueueWorkerModel.findById(WORKER_ID).lean();
+		expect(row?.mode).toBe("primary");
+		await scheduler2.shutdown();
+	});
+});

server/test/unit/services/dbQueueWorker.test.ts

@@ -79,6 +79,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses
 		updateScheduleById: jest.fn<any>().mockResolvedValue(true),
 		findAll: jest.fn<any>().mockResolvedValue([]),
 		findPage: jest.fn<any>().mockResolvedValue({ jobs: [], count: 0 }),
+		countDueBacklog: jest.fn<any>().mockResolvedValue(0),
 	};
 	const monitorsRepository = {
 		findByIds: jest.fn<any>().mockResolvedValue([makeMonitor()]),
@@ -186,7 +187,7 @@ describe("DBQueueWorker", () => {
 	describe("init", () => {
 		it("registers the worker in the heartbeat registry on startup", async () => {
 			const { mocks } = await start({ queueMode: "worker" });
-			expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker");
+			expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", true);
 		});
 
 		it("primary mode reconciles: seeds a check row per monitor plus the two cleanup rows", async () => {
@@ -263,7 +264,7 @@ describe("DBQueueWorker", () => {
 			active = worker;
 			await jest.advanceTimersByTimeAsync(1);
 
-			expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker");
+			expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", true);
 		});
 	});
 
@@ -768,6 +769,29 @@ describe("DBQueueWorker", () => {
 		});
 	});
 
+	// ── metrics accessors ───────────────────────────────────────────────────────────
+
+	describe("metrics accessors", () => {
+		it("countDueBacklog delegates to the repo with the current time", async () => {
+			const { worker, mocks } = await start({ queueMode: "worker" });
+			mocks.jobsRepository.countDueBacklog.mockResolvedValue(7);
+
+			expect(await worker.countDueBacklog()).toBe(7);
+			expect(mocks.jobsRepository.countDueBacklog).toHaveBeenCalledWith(expect.any(Number));
+		});
+
+		it("countAliveWorkers counts only recently-seen nodes that process jobs", async () => {
+			const { worker, mocks } = await start({ queueMode: "worker" });
+			mocks.queueWorkersRepository.findRecent.mockResolvedValue([
+				{ processesJobs: true },
+				{ processesJobs: true },
+				{ processesJobs: false }, // scheduler-only primary: alive but not a processor
+			]);
+
+			expect(await worker.countAliveWorkers()).toBe(2);
+		});
+	});
+
 	// ── flushQueues ───────────────────────────────────────────────────────────────
 
 	describe("flushQueues", () => {
@@ -799,7 +823,10 @@ describe("DBQueueWorker", () => {
 			const { worker, mocks } = createWorker();
 			active = worker;
 			mocks.jobsRepository.findAll.mockResolvedValue(rows);
-			mocks.queueWorkersRepository.findRecent.mockResolvedValue([{ workerId: "w" }]);
+			mocks.queueWorkersRepository.findRecent.mockResolvedValue([
+				{ workerId: "w", processesJobs: true },
+				{ workerId: "scheduler", processesJobs: false }, // scheduler-only primary excluded from metrics
+			]);
 
 			const metrics = await worker.getMetrics();
 
@@ -810,7 +837,7 @@ describe("DBQueueWorker", () => {
 			expect(metrics.failingJobs).toBe(1);
 			expect(metrics.jobsWithFailures).toHaveLength(1);
 			expect(metrics.jobsWithFailures[0]).toEqual(expect.objectContaining({ failCount: 3, failReason: "boom" }));
-			expect(metrics.workers).toEqual([{ workerId: "w" }]);
+			expect(metrics.workers).toEqual([{ workerId: "w", processesJobs: true }]);
 		});
 	});
 

server/test/unit/services/healthServer.test.ts

@@ -26,13 +26,13 @@ const freePort = (): Promise<number> =>
 		});
 	});
 
-const get = (port: number, path: string): Promise<{ status: number; body: string }> =>
+const get = (port: number, path: string): Promise<{ status: number; body: string; contentType: string }> =>
 	new Promise((resolve, reject) => {
 		http
 			.get({ host: "127.0.0.1", port, path }, (res) => {
 				let body = "";
 				res.on("data", (chunk) => (body += chunk));
-				res.on("end", () => resolve({ status: res.statusCode ?? 0, body }));
+				res.on("end", () => resolve({ status: res.statusCode ?? 0, body, contentType: String(res.headers["content-type"] ?? "") }));
 			})
 			.on("error", reject);
 	});
@@ -58,9 +58,16 @@ describe("HealthServer", () => {
 		active = null;
 	});
 
-	const setup = async (health: WorkerHealth) => {
+	// The metrics endpoint also calls countDueBacklog()/countAliveWorkers(); default them
+	// to 0 and let tests override (including to a rejection, to exercise the 500 path).
+	const setup = async (health: WorkerHealth, workerOverrides: Record<string, unknown> = {}) => {
 		const port = await freePort();
-		const worker = { getHealth: () => health } as any;
+		const worker = {
+			getHealth: () => health,
+			countDueBacklog: async () => 0,
+			countAliveWorkers: async () => 0,
+			...workerOverrides,
+		} as any;
 		const server = new HealthServer(createMockLogger(), String(port), worker);
 		await server.listen();
 		active = server;
@@ -133,4 +140,36 @@ describe("HealthServer", () => {
 			expect((await get(port, "/livez?probe=1")).status).toBe(200);
 		});
 	});
+
+	describe("/metrics", () => {
+		it("200 with prometheus content-type and all four gauges", async () => {
+			const { port } = await setup(makeHealth({ inFlight: 3, draining: false }), {
+				countDueBacklog: async () => 5,
+				countAliveWorkers: async () => 2,
+			});
+			const res = await get(port, "/metrics");
+
+			expect(res.status).toBe(200);
+			expect(res.contentType).toMatch(/text\/plain/);
+			expect(res.body).toContain("# TYPE checkmate_worker_due_backlog gauge");
+			expect(res.body).toContain("checkmate_worker_jobs_in_flight 3");
+			expect(res.body).toContain("checkmate_worker_due_backlog 5");
+			expect(res.body).toContain("checkmate_worker_alive_total 2");
+			expect(res.body).toContain("checkmate_worker_draining 0");
+		});
+
+		it("reports draining as 1 when the worker is draining", async () => {
+			const { port } = await setup(makeHealth({ draining: true }));
+			expect((await get(port, "/metrics")).body).toContain("checkmate_worker_draining 1");
+		});
+
+		it("500 when a metrics query fails", async () => {
+			const { port } = await setup(makeHealth(), {
+				countDueBacklog: async () => {
+					throw new Error("db down");
+				},
+			});
+			expect((await get(port, "/metrics")).status).toBe(500);
+		});
+	});
 });

server/test/unit/services/jobScheduler.test.ts

@@ -1,26 +1,124 @@
-import { describe, expect, it, jest } from "@jest/globals";
+import { describe, expect, it, jest, beforeEach, afterEach } from "@jest/globals";
 import { JobScheduler } from "../../../src/worker/worker.job-scheduler.ts";
+import type { Monitor } from "../../../src/domain/monitors/monitor.types.ts";
+import type { QueueMode } from "../../../src/domain/app-settings/app-settings.type.ts";
 import { createMockLogger } from "../../helpers/createMockLogger.ts";
 
 // ── Notes ──────────────────────────────────────────────────────────────────────
 //
 // JobScheduler is the scheduler-only primary base class: it owns the queue
-// registry/heartbeat but processes no jobs and has no buffer. drain() is therefore
-// a no-op beyond setting the stop flag — it must NOT tear down (clear timers or
-// deregister the worker); that stays in shutdown(). The real draining (wait for
-// in-flight, flush buffer) lives in DBQueueWorker and is covered separately.
+// registry/heartbeat AND, on a primary, the reconcile that seeds the jobs
+// collection — scheduling is its job whether or not it also processes. It runs no
+// polling loops and has no buffer, so drain() is a no-op beyond setting the stop
+// flag — it must NOT tear down (clear timers or deregister the worker); that stays
+// in shutdown(). The processing tier (poll loops, buffer flush) lives in
+// DBQueueWorker and is covered separately.
 
-const createScheduler = () => {
-	const jobsRepository = {} as any;
+const makeMonitor = (overrides?: Partial<Monitor>): Monitor =>
+	({
+		id: "m1",
+		teamId: "team",
+		type: "http",
+		interval: 60000,
+		status: "up",
+		isActive: true,
+		geoCheckEnabled: false,
+		geoCheckInterval: 300000,
+		lastEvaluatedAt: 0,
+		...overrides,
+	}) as Monitor;
+
+const createScheduler = (overrides?: { queueMode?: QueueMode; mocks?: Record<string, any> }) => {
+	const jobsRepository = {
+		upsertJob: jest.fn<any>().mockResolvedValue(true),
+		upsertCleanupJob: jest.fn<any>().mockResolvedValue(true),
+	};
 	const queueWorkersRepository = {
+		upsert: jest.fn<any>().mockResolvedValue(undefined),
 		deleteById: jest.fn<any>().mockResolvedValue(undefined),
 	};
+	const monitorsRepository = {
+		findAll: jest.fn<any>().mockResolvedValue([]),
+	};
 	const logger = createMockLogger();
-	const scheduler = new JobScheduler(jobsRepository, queueWorkersRepository as any, "worker-1");
-	return { scheduler, mocks: { jobsRepository, queueWorkersRepository, logger } };
+	const mocks = { jobsRepository, queueWorkersRepository, monitorsRepository, logger, ...overrides?.mocks };
+	const scheduler = new JobScheduler(
+		mocks.jobsRepository as any,
+		mocks.queueWorkersRepository as any,
+		mocks.monitorsRepository as any,
+		overrides?.queueMode ?? "primary",
+		mocks.logger as any,
+		"worker-1"
+	);
+	return { scheduler, mocks };
 };
 
 describe("JobScheduler", () => {
+	describe("init", () => {
+		beforeEach(() => {
+			jest.useFakeTimers();
+		});
+
+		afterEach(async () => {
+			jest.clearAllTimers();
+			jest.useRealTimers();
+		});
+
+		it("registers the worker in the heartbeat registry on startup", async () => {
+			const { scheduler, mocks } = createScheduler({ queueMode: "primary" });
+			await scheduler.init();
+			expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "primary", false);
+		});
+
+		it("primary mode reconciles: seeds a check row per monitor plus the two cleanup rows", async () => {
+			const { scheduler, mocks } = createScheduler({
+				queueMode: "primary",
+				mocks: { monitorsRepository: { findAll: jest.fn<any>().mockResolvedValue([makeMonitor()]) } },
+			});
+			await scheduler.init();
+
+			const seededTypes = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type);
+			expect(seededTypes).toContain("check");
+			const cleanupTypes = mocks.jobsRepository.upsertCleanupJob.mock.calls.map((c: any[]) => c[0].type);
+			expect(cleanupTypes).toContain("cleanup-orphaned");
+			expect(cleanupTypes).toContain("cleanup-retention");
+		});
+
+		it("primary mode also seeds a geo-check row for geo-enabled monitors", async () => {
+			const { scheduler, mocks } = createScheduler({
+				queueMode: "primary",
+				mocks: { monitorsRepository: { findAll: jest.fn<any>().mockResolvedValue([makeMonitor({ geoCheckEnabled: true })]) } },
+			});
+			await scheduler.init();
+
+			const seededTypes = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type);
+			expect(seededTypes).toContain("geo-check");
+		});
+
+		it("worker mode heartbeats but does not reconcile (no monitor reads, no seeds)", async () => {
+			const { scheduler, mocks } = createScheduler({
+				queueMode: "worker",
+				mocks: { monitorsRepository: { findAll: jest.fn<any>().mockResolvedValue([makeMonitor()]) } },
+			});
+			await scheduler.init();
+
+			expect(mocks.queueWorkersRepository.upsert).toHaveBeenCalledWith("worker-1", "worker", false);
+			expect(mocks.monitorsRepository.findAll).not.toHaveBeenCalled();
+			expect(mocks.jobsRepository.upsertJob).not.toHaveBeenCalled();
+			expect(mocks.jobsRepository.upsertCleanupJob).not.toHaveBeenCalled();
+		});
+
+		it("arms a heartbeat interval timer that shutdown() clears and deregisters", async () => {
+			const { scheduler, mocks } = createScheduler({ queueMode: "primary" });
+			await scheduler.init();
+			expect((scheduler as any).timers.has("heartbeat")).toBe(true);
+
+			await scheduler.shutdown();
+			expect((scheduler as any).timers.has("heartbeat")).toBe(false);
+			expect(mocks.queueWorkersRepository.deleteById).toHaveBeenCalledWith("worker-1");
+		});
+	});
+
 	describe("drain", () => {
 		it("sets the stop and draining flags without tearing down (no deregister)", async () => {
 			const { scheduler, mocks } = createScheduler();