add upsertCleanup to schedule immediate cleanup jobs, cleanup orphaned jobs

server/nodemon.json

@@ -1,5 +1,5 @@
 {
 	"ignore": ["node_modules/*"],
-	"watch": ["src/**/*.ts", "src/**/*.js", "*.json"],
+	"watch": ["src/**/*.ts", "src/**/*.js", "*.json", ".env"],
 	"ext": "ts,js,json"
 }

server/src/config/services.ts

@@ -285,6 +285,7 @@ export const initializeServices = async ({
 		checkService,
 		settingsService,
 		monitorsRepository,
+		jobsRepository,
 		teamsRepository,
 		monitorStatsRepository,
 		checksRepository,

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

@@ -43,6 +43,9 @@ export interface IJobsRepository {
 	// Create jobs. Takes a JobSeed
 	upsertJob(job: JobSeed): Promise<boolean>;
 
+	// Creates cleanup jobs, these are executed immediately so need their own method
+	upsertCleanupJob(job: JobSeed): Promise<boolean>;
+
 	// Pause/Resume
 	setActiveById(refId: string, isActive: boolean): Promise<boolean>;
 
@@ -55,6 +58,9 @@ export interface IJobsRepository {
 	// Delete a single job of type
 	deleteByIdAndType(refId: string, type: JobType): Promise<boolean>;
 
+	// Delete jobs that reference monitors that no longer exist
+	deleteByMonitorIdsNotIn(monitorIds: string[]): Promise<number>;
+
 	// Get job
 	findById(refId: string): Promise<Job[]>;
 

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

@@ -214,6 +214,31 @@ class MongoJobsRepository implements IJobsRepository {
 		return res.acknowledged;
 	};
 
+	upsertCleanupJob = async (job: JobSeed) => {
+		const res = await JobModel.updateOne(
+			{ _id: job.id },
+			{
+				$set: {
+					type: job.type,
+					refId: job.refId,
+					isActive: job.isActive,
+					intervalMs: job.intervalMs,
+					nextScheduledAt: job.nextScheduledAt,
+				},
+				$setOnInsert: {
+					lockedBy: null,
+					lockedUntil: null,
+					runCount: 0,
+					failCount: 0,
+					lastFinishedAt: null,
+					lastFailReason: null,
+				},
+			},
+			{ upsert: true }
+		);
+		return res.acknowledged;
+	};
+
 	setActiveById = async (refId: string, isActive: boolean) => {
 		const res = await JobModel.updateMany({ refId }, { $set: { isActive } }); // inverts check and geo check rows together
 		return res.modifiedCount > 0;
@@ -234,6 +259,12 @@ class MongoJobsRepository implements IJobsRepository {
 		return res.deletedCount > 0;
 	};
 
+	deleteByMonitorIdsNotIn = async (monitorIds: string[]): Promise<number> => {
+		// remove jobs who'se refId is not null (cleanup jobs) and not in the list of monitor IDs (orphaned monitor jobs)
+		const res = await JobModel.deleteMany({ refId: { $ne: null, $nin: monitorIds } });
+		return res.deletedCount ?? 0;
+	};
+
 	findById = async (refId: string) => {
 		const docs = await JobModel.find({ refId }).lean<JobDocument[]>();
 		return docs.map(this.toEntity);

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

@@ -274,8 +274,9 @@ export class DBQueueWorker implements IWorker {
 				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));
+		// Both cleanup jobs run immediately on every startup, then once a day
+		await this.jobsRepository.upsertCleanupJob(this.toCleanupJob("cleanup-orphaned", now));
+		await this.jobsRepository.upsertCleanupJob(this.toCleanupJob("cleanup-retention", now));
 	};
 
 	// Register worker

server/src/worker/worker.helper.ts

@@ -8,6 +8,7 @@ import { IGeoChecksRepository } from "@/domain/geo-checks/geo-check.repository.i
 import { IIncidentsRepository } from "@/domain/incidents/incident.repository.interface.js";
 import { IMonitorStatsRepository } from "@/domain/monitor-stats/monitor-stats.repository.interface.js";
 import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
+import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js";
 import { ITeamsRepository } from "@/domain/teams/team.repository.interface.js";
 import { ILogger } from "@/utils/logger.js";
 import { IReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js";
@@ -42,6 +43,7 @@ export class WorkerHelper implements IWorkerHelper {
 	private checkService: ICheckService;
 	private settingsService: ISettingsService;
 	private monitorsRepository: IMonitorsRepository;
+	private jobsRepository: IJobsRepository;
 	private teamsRepository: ITeamsRepository;
 	private monitorStatsRepository: IMonitorStatsRepository;
 	private checksRepository: IChecksRepository;
@@ -56,6 +58,7 @@ export class WorkerHelper implements IWorkerHelper {
 		checkService: ICheckService,
 		settingsService: ISettingsService,
 		monitorsRepository: IMonitorsRepository,
+		jobsRepository: IJobsRepository,
 		teamsRepository: ITeamsRepository,
 		monitorStatsRepository: IMonitorStatsRepository,
 		checksRepository: IChecksRepository,
@@ -69,6 +72,7 @@ export class WorkerHelper implements IWorkerHelper {
 		this.checkService = checkService;
 		this.settingsService = settingsService;
 		this.monitorsRepository = monitorsRepository;
+		this.jobsRepository = jobsRepository;
 		this.teamsRepository = teamsRepository;
 		this.monitorStatsRepository = monitorStatsRepository;
 		this.checksRepository = checksRepository;
@@ -190,6 +194,16 @@ export class WorkerHelper implements IWorkerHelper {
 					});
 				}
 
+				// Remove orphaned jobs
+				const deletedJobsCount = await this.jobsRepository.deleteByMonitorIdsNotIn(allMonitorIds);
+				if (deletedJobsCount > 0) {
+					this.logger.info({
+						message: `Deleted ${deletedJobsCount} orphaned jobs`,
+						service: SERVICE_NAME,
+						method: "getCleanupOrphanedJob",
+					});
+				}
+
 				this.logger.info({
 					message: "Cleanup of orphaned data completed",
 					service: SERVICE_NAME,

server/test/integration/jobsRepository.test.ts

@@ -347,6 +347,34 @@ describe("MongoJobsRepository", () => {
 			expect(row?.runCount).toBe(7);
 		});
 
+		it("upsertCleanupJob resets nextScheduledAt on an existing row so it is due again on restart", async () => {
+			const job: Job = {
+				id: "cleanup-orphaned",
+				type: "cleanup-orphaned",
+				refId: null,
+				isActive: true,
+				nextScheduledAt: NOW,
+				intervalMs: 86_400_000,
+				lockedBy: null,
+				lockedUntil: null,
+				runCount: 0,
+				failCount: 0,
+				lastFinishedAt: null,
+				lastFailReason: null,
+			};
+			await repo.upsertCleanupJob(job);
+
+			// row has already advanced a day into the future after a previous run
+			await JobModel.updateOne({ _id: "cleanup-orphaned" }, { $set: { nextScheduledAt: NOW + 86_400_000, runCount: 3 } });
+
+			// a restart reconcile re-upserts and pulls it due now
+			await repo.upsertCleanupJob({ ...job, nextScheduledAt: NOW + 1_000 });
+
+			const row = await readRow("cleanup-orphaned");
+			expect(row?.nextScheduledAt).toBe(NOW + 1_000); // schedule reset ($set)
+			expect(row?.runCount).toBe(3); // run history untouched ($setOnInsert)
+		});
+
 		it("setActiveById flips check and geo rows together", async () => {
 			await seedJob({ _id: "check:mon-1", type: "check" });
 			await seedJob({ _id: "geo:mon-1", type: "geo-check" });
@@ -378,6 +406,20 @@ describe("MongoJobsRepository", () => {
 			expect(await JobModel.countDocuments({ refId: "mon-1" })).toBe(0);
 		});
 
+		it("deleteByMonitorIdsNotIn drops jobs for missing monitors but keeps valid and global rows", async () => {
+			await seedJob({ _id: "check:keep", refId: "keep", type: "check" });
+			await seedJob({ _id: "geo:keep", refId: "keep", type: "geo-check" });
+			await seedJob({ _id: "check:gone", refId: "gone", type: "check" });
+			await seedJob({ _id: "cleanup-orphaned", refId: null, type: "cleanup-orphaned" });
+			await seedJob({ _id: "cleanup-retention", refId: null, type: "cleanup-retention" });
+
+			const deleted = await repo.deleteByMonitorIdsNotIn(["keep"]);
+
+			expect(deleted).toBe(1); // only check:gone
+			const remaining = (await JobModel.find().lean<JobDocument[]>()).map((d) => d._id).sort();
+			expect(remaining).toEqual(["check:keep", "cleanup-orphaned", "cleanup-retention", "geo:keep"]);
+		});
+
 		it("findById returns all of a monitor's rows", async () => {
 			await seedJob({ _id: "check:mon-1", type: "check" });
 			await seedJob({ _id: "geo:mon-1", type: "geo-check" });

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

@@ -65,6 +65,7 @@ const createWorker = (overrides?: { queueMode?: QueueMode; queuePrimaryProcesses
 	const jobsRepository = {
 		claimDueBatch: jest.fn<any>().mockResolvedValue([]),
 		upsertJob: jest.fn<any>().mockResolvedValue(true),
+		upsertCleanupJob: jest.fn<any>().mockResolvedValue(true),
 		upsertEvaluate: jest.fn<any>().mockResolvedValue(true),
 		recordSuccess: jest.fn<any>().mockResolvedValue(true),
 		recordFailure: jest.fn<any>().mockResolvedValue(true),
@@ -183,8 +184,9 @@ describe("DBQueueWorker", () => {
 
 			const seededTypes = mocks.jobsRepository.upsertJob.mock.calls.map((c: any[]) => c[0].type);
 			expect(seededTypes).toContain("check");
-			expect(seededTypes).toContain("cleanup-orphaned");
-			expect(seededTypes).toContain("cleanup-retention");
+			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 () => {
@@ -204,6 +206,7 @@ describe("DBQueueWorker", () => {
 		it("worker mode does not reconcile (never seeds jobs)", async () => {
 			const { mocks } = await start({ queueMode: "worker" });
 			expect(mocks.jobsRepository.upsertJob).not.toHaveBeenCalled();
+			expect(mocks.jobsRepository.upsertCleanupJob).not.toHaveBeenCalled();
 		});
 
 		it("starts a polling loop per job type, each claiming up to its concurrency cap", async () => {

server/test/unit/services/workerHelper.test.ts

@@ -17,6 +17,9 @@ const createHelper = (overrides?: Record<string, unknown>) => {
 		findAllMonitorIds: jest.fn().mockResolvedValue(["m1"]),
 		deleteByTeamIdsNotIn: jest.fn().mockResolvedValue(0),
 	};
+	const jobsRepository = {
+		deleteByMonitorIdsNotIn: jest.fn().mockResolvedValue(0),
+	};
 	const teamsRepository = {
 		findAllTeamIds: jest.fn().mockResolvedValue(["team"]),
 	};
@@ -48,6 +51,7 @@ const createHelper = (overrides?: Record<string, unknown>) => {
 		checkService,
 		settingsService,
 		monitorsRepository,
+		jobsRepository,
 		teamsRepository,
 		monitorStatsRepository,
 		checksRepository,
@@ -64,6 +68,7 @@ const createHelper = (overrides?: Record<string, unknown>) => {
 		defaults.checkService as any,
 		defaults.settingsService as any,
 		defaults.monitorsRepository as any,
+		defaults.jobsRepository as any,
 		defaults.teamsRepository as any,
 		defaults.monitorStatsRepository as any,
 		defaults.checksRepository as any,
@@ -170,6 +175,7 @@ describe("WorkerHelper", () => {
 			(defaults.checksRepository.deleteByMonitorIdsNotIn as jest.Mock).mockResolvedValue(4);
 			(defaults.incidentsRepository.deleteByMonitorIdsNotIn as jest.Mock).mockResolvedValue(1);
 			(defaults.geoChecksRepository.deleteByMonitorIdsNotIn as jest.Mock).mockResolvedValue(5);
+			(defaults.jobsRepository.deleteByMonitorIdsNotIn as jest.Mock).mockResolvedValue(6);
 
 			const job = helper.getCleanupOrphanedJob();
 			await job();
@@ -181,11 +187,13 @@ describe("WorkerHelper", () => {
 			expect(defaults.checksRepository.deleteByMonitorIdsNotIn).toHaveBeenCalledWith(["m1"]);
 			expect(defaults.incidentsRepository.deleteByMonitorIdsNotIn).toHaveBeenCalledWith(["m1"]);
 			expect(defaults.geoChecksRepository.deleteByMonitorIdsNotIn).toHaveBeenCalledWith(["m1"]);
+			expect(defaults.jobsRepository.deleteByMonitorIdsNotIn).toHaveBeenCalledWith(["m1"]);
 			expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("2 orphaned monitors") }));
 			expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("3 orphaned monitor stats") }));
 			expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("4 orphaned checks") }));
 			expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("1 orphaned incidents") }));
 			expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("5 orphaned geo checks") }));
+			expect(defaults.logger.info).toHaveBeenCalledWith(expect.objectContaining({ message: expect.stringContaining("6 orphaned jobs") }));
 		});
 
 		it("skips info logs when no orphaned data is found", async () => {