thread prefetched worker, remove dead code

server/src/worker/worker.check-evaluator.ts

@@ -1,11 +1,12 @@
 import { MonitorStatusResponse } from "@/types/network.js";
 import { MonitorEvaluation } from "@/worker/worker.interface.js";
 import { Check } from "@/domain/checks/check.type.js";
+import { Monitor } from "@/domain/monitors/monitor.types.js";
 import { IMonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js";
 import { IStatusService } from "@/service/statusService.js";
 
 export interface ICheckEvaluator {
-	evaluate(status: MonitorStatusResponse, check: Check): Promise<MonitorEvaluation>;
+	evaluate(status: MonitorStatusResponse, check: Check, monitor: Monitor): Promise<MonitorEvaluation>;
 }
 
 export class CheckEvaluator implements ICheckEvaluator {
@@ -13,11 +14,11 @@ export class CheckEvaluator implements ICheckEvaluator {
 		private statusService: IStatusService,
 		private monitorStatusPolicy: IMonitorStatusPolicy
 	) {}
-	evaluate = async (status: MonitorStatusResponse, check: Check) => {
+	evaluate = async (status: MonitorStatusResponse, check: Check, monitor: Monitor) => {
 		// ****************************
 		// Step 3:  Evaluate and return result to reactors
 		// ****************************
-		const statusChangeResult = await this.statusService.updateMonitorStatus(status, check);
+		const statusChangeResult = await this.statusService.updateMonitorStatus(status, check, monitor);
 
 		// Step 5.  Get decisions and create an evaluation obj
 		const decision = this.monitorStatusPolicy.evaluate(statusChangeResult);

server/src/worker/worker.check-pipeline.ts

@@ -7,8 +7,6 @@ import { IBufferService } from "@/service/bufferService.js";
 import { AppError } from "@/utils/AppError.js";
 import { supportsGeoCheck } from "@/domain/monitors/monitor.types.js";
 import { IGeoChecksService } from "@/domain/geo-checks/geo-check.service.js";
-import { ICheckProducer } from "@/worker/worker.check-producer.js";
-import { ICheckEvaluator } from "@/worker/worker.check-evaluator.js";
 
 const SERVICE_NAME = "CheckPipeline";
 
@@ -16,22 +14,6 @@ export interface ICheckPipeline {
 	run(monitor: Monitor): Promise<MonitorEvaluation | null>; // null = skipped
 }
 
-export class CheckPipeline implements ICheckPipeline {
-	constructor(
-		private checkProducer: ICheckProducer,
-		private checkEvaluator: ICheckEvaluator
-	) {}
-
-	run = async (monitor: Monitor): Promise<MonitorEvaluation | null> => {
-		const checkResult = await this.checkProducer.produce(monitor);
-		if (!checkResult) {
-			return null; // skipped due to maintenance / geo guards / other guards
-		}
-		const evaluation = await this.checkEvaluator.evaluate(checkResult.status, checkResult.check);
-		return evaluation;
-	};
-}
-
 export class GeoChecksPipeline implements ICheckPipeline {
 	constructor(
 		private maintenanceWindowsRepository: IMaintenanceWindowsRepository,

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

@@ -151,7 +151,7 @@ export class DBQueueWorker implements IWorker {
 
 	private runCheck = async (job: Job) => {
 		if (!job.refId) return;
-		const [monitor] = await this.monitorsRepository.findByIds([job.refId]); // job row has no teamId
+		const monitor = await this.monitorsRepository.findByIdLean(job.refId); // job row has no teamId
 		if (!monitor) return;
 		await this.checkProducer.produce(monitor);
 	};
@@ -162,19 +162,22 @@ export class DBQueueWorker implements IWorker {
 
 	private runEvaluate = async (job: Job) => {
 		if (!job.refId) return;
-		const [monitor] = await this.monitorsRepository.findByIds([job.refId]); // job row has no teamId
+		const monitor = await this.monitorsRepository.findByIdLean(job.refId); // job row has no teamId
 		if (!monitor) return;
 		const checks = await this.checksRepository.findUnevaluatedByMonitorId(job.refId, monitor.lastEvaluatedAt);
+
+		let current = monitor;
 		for (const check of checks) {
 			const status = this.checkService.toStatusResponse(check);
-			const evaluation = await this.checkEvaluator.evaluate(status, check);
+			const evaluation = await this.checkEvaluator.evaluate(status, check, current);
 			await this.dispatcher.dispatch(evaluation); // Handle incidents and notifications
-			await this.monitorsRepository.updateById(job.refId, monitor.teamId, { lastEvaluatedAt: this.checkService.toLastEvaluatedAt(check) });
+			await this.monitorsRepository.updateById(job.refId, current.teamId, { lastEvaluatedAt: this.checkService.toLastEvaluatedAt(check) });
+			current = evaluation.statusChange.monitor; // fresh statusWindow/status/counters for the next check
 		}
 	};
 
 	private runGeoCheck = async (job: Job) => {
-		const [monitor] = await this.monitorsRepository.findByIds([job.refId!]);
+		const monitor = await this.monitorsRepository.findByIdLean(job.refId!);
 		if (monitor) await this.geoCheckPipeline.run(monitor); // returns null; no evaluate handoff
 	};
 

server/src/worker/worker.helper.ts

@@ -1,5 +1,4 @@
 const SERVICE_NAME = "JobQueueHelper";
-import type { Monitor } from "@/domain/monitors/monitor.types.js";
 import { ISettingsService } from "@/domain/app-settings/app-settings.service.js";
 import { ICheckService } from "@/domain/checks/check.service.js";
 import { CHECK_TTL_SENTINEL } from "@/domain/checks/check.type.js";
@@ -11,13 +10,9 @@ import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interf
 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";
-import { ICheckPipeline } from "@/worker/worker.check-pipeline.js";
 
 export interface IWorkerHelper {
 	readonly serviceName: string;
-	getHeartbeatJob(): (monitor: Monitor) => Promise<void>;
-	getHeartbeatGeoJob(): (monitor: Monitor) => Promise<void>;
 	getCleanupOrphanedJob(): () => Promise<void>;
 	getCleanupRetentionJob(): () => Promise<void>;
 }
@@ -49,9 +44,6 @@ export class WorkerHelper implements IWorkerHelper {
 	private checksRepository: IChecksRepository;
 	private incidentsRepository: IIncidentsRepository;
 	private geoChecksRepository: IGeoChecksRepository;
-	private reactorDispatcher: IReactorDispatcher;
-	private checkPipeline: ICheckPipeline;
-	private geoCheckPipeline: ICheckPipeline;
 
 	constructor(
 		logger: ILogger,
@@ -63,10 +55,7 @@ export class WorkerHelper implements IWorkerHelper {
 		monitorStatsRepository: IMonitorStatsRepository,
 		checksRepository: IChecksRepository,
 		incidentsRepository: IIncidentsRepository,
-		geoChecksRepository: IGeoChecksRepository,
-		reactorDispatcher: IReactorDispatcher,
-		checkPipeline: ICheckPipeline,
-		geoCheckPipeline: ICheckPipeline
+		geoChecksRepository: IGeoChecksRepository
 	) {
 		this.logger = logger;
 		this.checkService = checkService;
@@ -78,48 +67,12 @@ export class WorkerHelper implements IWorkerHelper {
 		this.checksRepository = checksRepository;
 		this.incidentsRepository = incidentsRepository;
 		this.geoChecksRepository = geoChecksRepository;
-		this.reactorDispatcher = reactorDispatcher;
-		this.checkPipeline = checkPipeline;
-		this.geoCheckPipeline = geoCheckPipeline;
 	}
 
 	get serviceName() {
 		return WorkerHelper.SERVICE_NAME;
 	}
 
-	getHeartbeatJob = () => {
-		return async (monitor: Monitor) => {
-			try {
-				const evaluation = await this.checkPipeline.run(monitor);
-				if (evaluation) await this.reactorDispatcher.dispatch(evaluation);
-			} catch (error: unknown) {
-				this.logger.warn({
-					message: error instanceof Error ? error.message : "Unknown error",
-					service: SERVICE_NAME,
-					method: "getMonitorJob",
-					stack: error instanceof Error ? error.stack : undefined,
-				});
-				throw error;
-			}
-		};
-	};
-
-	getHeartbeatGeoJob = () => {
-		return async (monitor: Monitor) => {
-			try {
-				await this.geoCheckPipeline.run(monitor);
-			} catch (error: unknown) {
-				this.logger.error({
-					message: error instanceof Error ? error.message : "Unknown error",
-					service: SERVICE_NAME,
-					method: "getHeartbeatGeoJob",
-					stack: error instanceof Error ? error.stack : undefined,
-				});
-				// Don't throw - geo check failures shouldn't crash the job scheduler
-			}
-		};
-	};
-
 	getCleanupOrphanedJob = () => {
 		return async () => {
 			try {