dbQueueWorker

server/scripts/seed-monitors.js

@@ -1,6 +1,6 @@
 import mongoose from "mongoose";
-import { MonitorModel } from "../dist/db/models/Monitor.js";
-import { UserModel } from "../dist/db/models/User.js";
+import { MonitorModel } from "../dist/domain/monitors/monitor.model.js";
+import { UserModel } from "../dist/domain/users/user.model.js";
 
 // Seeds a large number of HTTP monitors against real, resolving domains so the
 // scheduler can be exercised under load. Run after `npm run build` (imports the
@@ -241,6 +241,7 @@ async function run() {
 			url: `https://${domain}`,
 			isActive: true,
 			interval,
+			lastEvaluatedAt: 0, // evaluator watermark; without it findUnevaluatedByMonitorId gets new Date(undefined)
 		});
 
 		if (docs.length === batchSize) {

server/src/config/services.ts

@@ -109,6 +109,8 @@ import { IncidentReactor } from "@/worker/reactors/reactor.incident.js";
 import { CheckPipeline, GeoChecksPipeline } from "@/worker/worker.check-pipeline.js";
 import { CheckProducer } from "@/worker/worker.check-producer.js";
 import { CheckEvaluator } from "@/worker/worker.check-evaluator.js";
+import { DBQueueWorker } from "@/worker/worker.db-queue.js";
+import MongoJobsRepository from "@/domain/jobs/job.repository.mongo.js";
 
 export type InitializedServices = {
 	settingsService: ISettingsService;
@@ -233,6 +235,7 @@ export const initializeServices = async ({
 	const teamsRepository = new MongoTeamsRepository();
 	const maintenanceWindowsRepository = new MongoMaintenanceWindowsRepository();
 	const queueWorkersRepository = new MongoQueueWorkersRepository();
+	const jobsRepository = new MongoJobsRepository();
 
 	// Inject settings repository into settings service (now that DB is connected)
 	(settingsService as SettingsService).setRepository(settingsRepository);
@@ -360,6 +363,21 @@ export const initializeServices = async ({
 		case "lessSimpleQueue":
 			worker = await LessSimpleWorker.create(logger, workerHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode);
 			break;
+		case "dbQueue":
+			worker = await DBQueueWorker.create(
+				logger,
+				jobsRepository,
+				monitorsRepository,
+				checksRepository,
+				checkService,
+				checkProducer,
+				checkEvaluator,
+				geoCheckPipeline,
+				reactorDispatcher,
+				workerHelper
+			);
+			break;
+
 		default:
 			worker = await SuperSimpleQueue.create(logger, workerHelper, monitorsRepository);
 	}

server/src/domain/app-settings/app-settings.type.ts

@@ -1,7 +1,7 @@
 export const DbTypes = ["mongodb"] as const;
 export type DbType = (typeof DbTypes)[number];
 
-export const QueueTypes = ["lessSimpleQueue", "superSimpleQueue"] as const;
+export const QueueTypes = ["lessSimpleQueue", "superSimpleQueue", "dbQueue"] as const;
 export type QueueType = (typeof QueueTypes)[number];
 
 export const QueueModes = ["primary", "worker"] as const;

server/src/domain/checks/check.service.ts

@@ -11,6 +11,7 @@ const SERVICE_NAME = "checkService";
 
 export interface ICheckService {
 	createChecks(checks: Check[]): Promise<Check[]>;
+	toLastEvaluatedAt(check: Check): number;
 	toCheck(statusResponse: MonitorStatusResponse<MonitorPayloadMap[keyof MonitorPayloadMap]>): Check | undefined;
 	toStatusResponse(check: Check): MonitorStatusResponse<MonitorPayloadMap[keyof MonitorPayloadMap]>;
 	getChecksByMonitor(params: {
@@ -57,6 +58,10 @@ export class CheckService implements ICheckService {
 		return this.checksRepository.createChecks(checks);
 	};
 
+	toLastEvaluatedAt = (check: Check) => {
+		return new Date(check.createdAt).getTime();
+	};
+
 	toCheck = (statusResponse: MonitorStatusResponse<MonitorPayloadMap[keyof MonitorPayloadMap]>): Check | undefined => {
 		const { monitorId, teamId, type, status, responseTime, code, message, payload, timings } = statusResponse;
 

server/src/domain/jobs/job.repository.interface.ts

@@ -1,4 +1,4 @@
-import { Job, JobType } from "@/domain/jobs/job.type.js";
+import { Job, JobSeed, JobType } from "@/domain/jobs/job.type.js";
 
 export type JobPageQuery = {
 	page?: number;
@@ -23,6 +23,8 @@ export interface IJobsRepository {
 	// Record failure, release lease, bump fail count, record reason, reset nextScheduledAt for retry
 	recordFailure(id: string, error: unknown, now: number): Promise<boolean>;
 
+	recordOneShot(id: string, now: number): Promise<boolean>;
+
 	// ********************
 	// Hand off to evaluator // This is accomplished by creating an "evaluation" type job
 	// ********************
@@ -32,8 +34,8 @@ export interface IJobsRepository {
 	// Job CRUD
 	// ********************
 
-	// Create jobs
-	upsertJob(job: Job): Promise<boolean>;
+	// Create jobs. Takes a JobSeed — the repo owns volatile lease/counter defaults on insert.
+	upsertJob(job: JobSeed): Promise<boolean>;
 
 	// Pause/Resume
 	setActiveById(refId: string, isActive: boolean): Promise<boolean>;
@@ -44,6 +46,9 @@ export interface IJobsRepository {
 	// Delete, drop all rows
 	deleteById(refId: string): Promise<boolean>;
 
+	// Delete a single typed row for a monitor (e.g. drop just the geo row, keep check/evaluate)
+	deleteByIdAndType(refId: string, type: JobType): Promise<boolean>;
+
 	// Get job
 	findById(refId: string): Promise<Job[]>;
 

server/src/domain/jobs/job.repository.mongo.ts

@@ -1,6 +1,6 @@
 import { IJobsRepository, JobPageQuery } from "@/domain/jobs/job.repository.interface.js";
 import JobModel, { JobDocument } from "@/domain/jobs/job.model.js";
-import { JobType, Job, BACKOFF_MS, LOCK_MS } from "@/domain/jobs/job.type.js";
+import { JobType, Job, JobSeed, BACKOFF_MS, LOCK_MS, PARKED, jobId } from "@/domain/jobs/job.type.js";
 import { hostname } from "node:os";
 import { randomUUID } from "node:crypto";
 
@@ -98,11 +98,31 @@ class MongoJobsRepository implements IJobsRepository {
 		return res.modifiedCount === 1;
 	};
 
+	recordOneShot = async (id: string, now: number) => {
+		const res = await JobModel.updateOne(
+			{
+				_id: id,
+			},
+			{
+				$set: {
+					nextScheduledAt: PARKED,
+					lockedBy: null, // Remove lock
+					lockedUntil: null,
+					lastFinishedAt: now,
+				},
+				$inc: {
+					runCount: 1, // Increment run count
+				},
+			}
+		);
+		return res.modifiedCount === 1;
+	};
+
 	upsertEvaluate = async (monitorId: string, now: number) => {
 		const filter = { type: "evaluate", refId: monitorId };
 		const update = {
 			$setOnInsert: {
-				_id: `evaluate:${monitorId}`,
+				_id: jobId("evaluate", monitorId),
 				// refId and Type inferred from filter
 				isActive: true,
 				intervalMs: null,
@@ -130,7 +150,7 @@ class MongoJobsRepository implements IJobsRepository {
 			throw error;
 		}
 	};
-	upsertJob = async (job: Job) => {
+	upsertJob = async (job: JobSeed) => {
 		const res = await JobModel.updateOne(
 			{ _id: job.id },
 			{
@@ -172,6 +192,11 @@ class MongoJobsRepository implements IJobsRepository {
 		return res.deletedCount > 0;
 	};
 
+	deleteByIdAndType = async (refId: string, type: JobType) => {
+		const res = await JobModel.deleteOne({ refId, type }); // Delete a single typed row, e.g. just the geo row
+		return res.deletedCount > 0;
+	};
+
 	findById = async (refId: string) => {
 		const docs = await JobModel.find({ refId }).lean<JobDocument[]>();
 		return docs.map(this.toEntity);

server/src/domain/jobs/job.type.ts

@@ -2,6 +2,10 @@ export const JOB_TYPES = ["check", "geo-check", "evaluate", "cleanup-orphaned",
 export type JobType = (typeof JOB_TYPES)[number];
 export const LOCK_MS = 30_000;
 export const BACKOFF_MS = 5_000;
+export const PARKED = Number.MAX_SAFE_INTEGER;
+
+// Canonical _id scheme: monitor-bound rows are `${type}:${refId}`; global rows (refId null) are just the type.
+export const jobId = (type: JobType, refId: string | null): string => (refId === null ? type : `${type}:${refId}`);
 
 export type Job = {
 	id: string;
@@ -23,3 +27,6 @@ export type Job = {
 	lastFinishedAt: number | null;
 	lastFailReason: string | null;
 };
+
+// These are the only values to set when creating a job.  Repo owns the rest
+export type JobSeed = Pick<Job, "id" | "type" | "refId" | "isActive" | "intervalMs" | "nextScheduledAt">;

server/src/worker/worker.db-queue.ts

@@ -0,0 +1,324 @@
+import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js";
+import { IWorker, WorkerJobsPagination, WorkerJobSummary, WorkerMetrics } from "@/worker/worker.interface.js";
+import { type Job, type JobSeed, type JobType, jobId } from "@/domain/jobs/job.type.js";
+import { randomUUID } from "node:crypto";
+import { hostname } from "node:os";
+import { ILogger } from "@/utils/logger.js";
+import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js";
+import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
+import { IChecksRepository } from "@/domain/checks/check.repository.interface.js";
+import { ICheckService } from "@/domain/checks/check.service.js";
+import { ICheckProducer } from "@/worker/worker.check-producer.js";
+import { ICheckEvaluator } from "@/worker/worker.check-evaluator.js";
+import { ICheckPipeline } from "@/worker/worker.check-pipeline.js";
+import { IReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js";
+import { IWorkerHelper } from "@/worker/worker.helper.js";
+const SERVICE_NAME = "JobQueue";
+const POLL_MS = 250;
+const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000;
+const CONCURRENCY: Record<JobType, number> = {
+	check: 50,
+	"geo-check": 20,
+	evaluate: 20,
+	"cleanup-orphaned": 1,
+	"cleanup-retention": 1,
+};
+
+export class DBQueueWorker implements IWorker {
+	static SERVICE_NAME = SERVICE_NAME;
+	private readonly workerId = `${hostname()}:${process.pid}:${randomUUID().slice(0, 8)}`;
+	private stopped = false;
+	private timers = new Map<JobType, NodeJS.Timeout>();
+	private inFlight: Record<JobType, number> = {
+		check: 0,
+		"geo-check": 0,
+		evaluate: 0,
+		"cleanup-orphaned": 0,
+		"cleanup-retention": 0,
+	};
+
+	constructor(
+		private logger: ILogger,
+		private jobsRepository: IJobsRepository,
+		private monitorsRepository: IMonitorsRepository,
+		private checksRepository: IChecksRepository,
+		private checkService: ICheckService, // toStatusResponse + toLastEvaluatedAt
+		private checkProducer: ICheckProducer,
+		private checkEvaluator: ICheckEvaluator,
+		private geoCheckPipeline: ICheckPipeline,
+		private dispatcher: IReactorDispatcher,
+		private helper: IWorkerHelper
+	) {}
+
+	get serviceName() {
+		return DBQueueWorker.SERVICE_NAME;
+	}
+
+	static async create(
+		logger: ILogger,
+		jobsRepository: IJobsRepository,
+		monitorsRepository: IMonitorsRepository,
+		checksRepository: IChecksRepository,
+		checkService: ICheckService,
+		checkProducer: ICheckProducer,
+		checkEvaluator: ICheckEvaluator,
+		geoCheckPipeline: ICheckPipeline,
+		dispatcher: IReactorDispatcher,
+		helper: IWorkerHelper
+	): Promise<DBQueueWorker> {
+		const instance = new DBQueueWorker(
+			logger,
+			jobsRepository,
+			monitorsRepository,
+			checksRepository,
+			checkService,
+			checkProducer,
+			checkEvaluator,
+			geoCheckPipeline,
+			dispatcher,
+			helper
+		);
+		await instance.init();
+		return instance;
+	}
+
+	// Helpers, builds job seeds
+	private toCheckJob = (monitor: Monitor, now: number): JobSeed => ({
+		id: jobId("check", monitor.id),
+		type: "check",
+		refId: monitor.id,
+		isActive: monitor.isActive,
+		nextScheduledAt: now + Math.floor(Math.random() * monitor.interval), // Avoid herd
+		intervalMs: monitor.interval,
+	});
+
+	private toGeoCheckJob = (monitor: Monitor, now: number): JobSeed => ({
+		...this.toCheckJob(monitor, now),
+		id: jobId("geo-check", monitor.id),
+		type: "geo-check",
+		intervalMs: monitor.geoCheckInterval!,
+	});
+
+	private toCleanupJob = (type: "cleanup-orphaned" | "cleanup-retention", now: number): JobSeed => ({
+		id: jobId(type, null),
+		type,
+		refId: null,
+		isActive: true,
+		nextScheduledAt: now,
+		intervalMs: CLEANUP_INTERVAL_MS,
+	});
+
+	private toSummary = (job: Job): WorkerJobSummary => ({
+		monitorId: job.refId ?? job.id,
+		monitorUrl: null, // Jobs don't have URL
+		monitorType: job.type,
+		monitorInterval: job.intervalMs,
+		monitorGeoInterval: null,
+		monitorActive: job.isActive,
+		active: job.isActive,
+		lockedBy: job.lockedBy,
+		lockedUntil: job.lockedUntil,
+		lockedAt: null, // no lockedAt
+		runCount: job.runCount,
+		failCount: job.failCount,
+		failReason: job.lastFailReason,
+		lastRunAt: null, // not tracked
+		lastFinishedAt: job.lastFinishedAt,
+		lastRunTook: null, // not tracked
+		lastFailedAt: null, // not tracked
+		repeat: job.intervalMs,
+	});
+
+	// ********************
+	// Stage 1:  This can eventually be offloaded to a separate service
+	// ********************
+
+	private runCheck = async (job: Job) => {
+		if (!job.refId) return;
+		const [monitor] = await this.monitorsRepository.findByIds([job.refId]); // job row has no teamId
+		if (!monitor) return;
+		const produced = await this.checkProducer.produce(monitor);
+		if (produced) {
+			await this.jobsRepository.upsertEvaluate(monitor.id, Date.now());
+		}
+	};
+
+	// ********************
+	// Stage 2:  Evaluator loop stays here
+	// ********************
+
+	private runEvaluate = async (job: Job) => {
+		if (!job.refId) return;
+		const [monitor] = await this.monitorsRepository.findByIds([job.refId]); // job row has no teamId
+		if (!monitor) return;
+		const checks = await this.checksRepository.findUnevaluatedByMonitorId(job.refId, monitor.lastEvaluatedAt);
+		for (const check of checks) {
+			const status = this.checkService.toStatusResponse(check);
+			const evaluation = await this.checkEvaluator.evaluate(status, check);
+			await this.dispatcher.dispatch(evaluation); // Handle incidents and notifications
+			await this.monitorsRepository.updateById(job.refId, monitor.teamId, { lastEvaluatedAt: this.checkService.toLastEvaluatedAt(check) });
+		}
+	};
+
+	private runGeoCheck = async (job: Job) => {
+		const [monitor] = await this.monitorsRepository.findByIds([job.refId!]);
+		if (monitor) await this.geoCheckPipeline.run(monitor); // returns null; no evaluate handoff
+	};
+
+	private runJob = async (job: Job) => {
+		try {
+			switch (job.type) {
+				case "check":
+					await this.runCheck(job);
+					break;
+				case "evaluate":
+					await this.runEvaluate(job);
+					break;
+				case "geo-check":
+					await this.runGeoCheck(job);
+					break;
+				case "cleanup-orphaned":
+					await this.helper.getCleanupOrphanedJob()(); // Get job and execute
+					break;
+				case "cleanup-retention":
+					await this.helper.getCleanupRetentionJob()(); // Get job and execute
+					break;
+			}
+
+			if (job.intervalMs === null) {
+				// One-shot job
+				await this.jobsRepository.recordOneShot(job.id, Date.now());
+			} else {
+				await this.jobsRepository.recordSuccess(job.id, job.nextScheduledAt, job.intervalMs, Date.now());
+			}
+		} catch (error: unknown) {
+			await this.jobsRepository.recordFailure(job.id, error, Date.now()); // Record failure and release lock
+			this.logger.warn({
+				message: error instanceof Error ? error.message : String(error),
+				service: SERVICE_NAME,
+				method: `runJob:${job.type}`,
+			});
+		}
+	};
+
+	private startLoop = (type: JobType) => {
+		const tick = async () => {
+			if (this.stopped) return;
+			try {
+				// Drain rows up to concurrency limit
+				while (this.inFlight[type] < CONCURRENCY[type]) {
+					const job = await this.jobsRepository.claimDue(type, Date.now()); // Claim a job
+					if (!job) break; // we're done, no jobs due
+					this.inFlight[type]++; // increment in flight jobs to stay within concurrency limits
+					this.runJob(job).finally(() => {
+						this.inFlight[type]--; // job is done, decrement in flight count
+					});
+				}
+			} catch (error: unknown) {
+				this.logger.error({
+					message: error instanceof Error ? error.message : String(error),
+					service: SERVICE_NAME,
+					method: `loop:${type}`,
+				});
+			}
+			if (!this.stopped) this.timers.set(type, setTimeout(tick, POLL_MS)); // re-arm unless shutting down
+		};
+		tick();
+	};
+
+	// Seed queue
+	private reconcile = async () => {
+		const now = Date.now();
+		const monitors = (await this.monitorsRepository.findAll()) ?? [];
+		for (const monitor of monitors) {
+			await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now));
+			if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) {
+				await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, now));
+			}
+		}
+		await this.jobsRepository.upsertJob(this.toCleanupJob("cleanup-orphaned", now));
+		await this.jobsRepository.upsertJob(this.toCleanupJob("cleanup-retention", now));
+	};
+
+	init = async () => {
+		this.stopped = false;
+		await this.reconcile();
+		for (const type of Object.keys(this.inFlight) as JobType[]) {
+			this.startLoop(type);
+		}
+		return true;
+	};
+	addJob = async (_monitorId: string, monitor: Monitor) => {
+		const now = Date.now();
+		await this.jobsRepository.upsertJob(this.toCheckJob(monitor, now));
+		if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, now));
+	};
+	deleteJob = async (monitor: Monitor) => {
+		await this.jobsRepository.deleteById(monitor.id);
+	};
+	pauseJob = async (monitor: Monitor) => {
+		await this.jobsRepository.setActiveById(monitor.id, false);
+	};
+	resumeJob = async (monitor: Monitor) => {
+		await this.jobsRepository.setActiveById(monitor.id, true);
+	};
+	updateJob = async (monitor: Monitor) => {
+		await this.jobsRepository.updateScheduleById(monitor.id, "check", monitor.interval);
+		if (supportsGeoCheck(monitor.type) && monitor.geoCheckEnabled) {
+			await this.jobsRepository.upsertJob(this.toGeoCheckJob(monitor, Date.now()));
+		} else {
+			await this.jobsRepository.deleteByIdAndType(monitor.id, "geo-check");
+		}
+	};
+	shutdown = async () => {
+		this.stopped = true;
+		// Stop all timers
+		for (const timer of this.timers.values()) {
+			clearTimeout(timer);
+		}
+		this.timers.clear(); // Clear the map out
+	};
+	getMetrics = async (): Promise<WorkerMetrics> => {
+		const rows = await this.jobsRepository.findAll();
+		const now = Date.now();
+		return rows.reduce<WorkerMetrics>(
+			(acc, job) => {
+				acc.jobs++;
+				acc.totalRuns += job.runCount;
+				acc.totalFailures += job.failCount;
+				if (job.lockedBy !== null && job.lockedUntil !== null && job.lockedUntil > now) acc.activeJobs++;
+				if (job.failCount > 0) {
+					acc.failingJobs++;
+					acc.jobsWithFailures.push({
+						monitorId: job.refId ?? job.id,
+						monitorUrl: null,
+						monitorType: job.type,
+						failedAt: job.lastFinishedAt,
+						failCount: job.failCount,
+						failReason: job.lastFailReason,
+					});
+				}
+				return acc;
+			},
+			{
+				jobs: 0,
+				activeJobs: 0,
+				failingJobs: 0,
+				jobsWithFailures: [],
+				totalRuns: 0,
+				totalFailures: 0,
+				workers: [{ workerId: this.workerId, mode: "primary", lastSeenAt: now }],
+			}
+		);
+	};
+
+	getJobs = async (pagination: WorkerJobsPagination) => {
+		const { jobs, count } = await this.jobsRepository.findPage(pagination);
+		return { jobs: jobs.map(this.toSummary), count };
+	};
+	flushQueues = async () => {
+		await this.shutdown();
+		const ok = await this.init();
+		return { success: ok };
+	};
+}