extract reactors and pipeline

server/src/config/services.ts

@@ -28,12 +28,13 @@ import { IUserService, UserService } from "@/domain/users/user.service.js";
 import { BufferService, IBufferService } from "@/service/bufferService.js";
 import { EmailService, IEmailService } from "@/service/emailService.js";
 import { GlobalPingService } from "@/service/globalPingService.js";
-import { QueueHelper } from "@/worker/worker.helper.js";
+import { WorkerHelper } from "@/worker/worker.helper.js";
 import { IWorker } from "@/worker/worker.interface.js";
 import { LessSimpleWorker } from "@/worker/worker.less-simple.js";
 import { SuperSimpleQueue } from "@/worker/worker.super-simple.js";
 import { INetworkService, NetworkService } from "@/service/networkService.js";
 import { IStatusService, StatusService } from "@/service/statusService.js";
+import { MonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js";
 
 // Network providers
 import { PingProvider } from "@/service/network/PingProvider.js";
@@ -102,6 +103,10 @@ import MongoUsersRepository from "@/domain/users/user.repository.mongo.js";
 import { ILogger } from "@/utils/logger.js";
 import { AppError } from "@/utils/AppError.js";
 import { type QueueMode } from "@/domain/app-settings/app-settings.type.js";
+import { NotificationReactor } from "@/worker/reactors/reactor.notification.js";
+import { ReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js";
+import { IncidentReactor } from "@/worker/reactors/reactor.incident.js";
+import { CheckPipeline, GeoChecksPipeline } from "@/worker/worker.check-pipeline.js";
 
 export type InitializedServices = {
 	settingsService: ISettingsService;
@@ -309,23 +314,39 @@ export const initializeServices = async ({
 
 	const tagsService = new TagsService(tagsRepository, monitorsRepository);
 
-	const queueHelper = new QueueHelper(
-		logger,
+	const monitorStatusPolicy = new MonitorStatusPolicy();
+	// reactors and dispatcher
+	const notificationReactor = new NotificationReactor(notificationsService);
+	const incidentReactor = new IncidentReactor(incidentService);
+	const reactorDispatcher = new ReactorDispatcher(logger, [notificationReactor, incidentReactor]);
+
+	// pipelines
+	const checkPipeline = new CheckPipeline(
+		monitorsRepository,
+		maintenanceWindowsRepository,
+		checkService,
 		networkService,
+		bufferService,
+		monitorStatusPolicy,
 		statusService,
-		notificationsService,
+		logger
+	);
+
+	const geoCheckPipeline = new GeoChecksPipeline(maintenanceWindowsRepository, geoChecksService, bufferService, logger);
+
+	const workerHelper = new WorkerHelper(
+		logger,
 		checkService,
 		settingsService,
-		bufferService,
-		incidentService,
-		maintenanceWindowsRepository,
 		monitorsRepository,
 		teamsRepository,
 		monitorStatsRepository,
 		checksRepository,
 		incidentsRepository,
-		geoChecksService,
-		geoChecksRepository
+		geoChecksRepository,
+		reactorDispatcher,
+		checkPipeline,
+		geoCheckPipeline
 	);
 
 	if (queueMode === "worker" && envSettings.queueType !== "lessSimpleQueue") {
@@ -340,10 +361,10 @@ export const initializeServices = async ({
 	let worker: IWorker;
 	switch (envSettings.queueType) {
 		case "lessSimpleQueue":
-			worker = await LessSimpleWorker.create(logger, queueHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode);
+			worker = await LessSimpleWorker.create(logger, workerHelper, monitorsRepository, queueWorkersRepository, envSettings, queueMode);
 			break;
 		default:
-			worker = await SuperSimpleQueue.create(logger, queueHelper, monitorsRepository);
+			worker = await SuperSimpleQueue.create(logger, workerHelper, monitorsRepository);
 	}
 
 	// Business services

server/src/worker/reactors/reactor.dispatcher.ts

@@ -0,0 +1,39 @@
+import { ILogger } from "@/utils/logger.js";
+import { MonitorEvaluation } from "@/worker/worker.interface.js";
+import { IMonitorReactor } from "@/worker/reactors/reactor.interface.js";
+
+const SERVICE_NAME = "ReactorDispatcher";
+
+export interface IReactorDispatcher {
+	dispatch(evaluation: MonitorEvaluation): Promise<void>;
+}
+
+export class ReactorDispatcher implements IReactorDispatcher {
+	constructor(
+		private logger: ILogger,
+		private reactors: IMonitorReactor[]
+	) {}
+
+	private logFailure(reactor: IMonitorReactor, evaluation: MonitorEvaluation, error: unknown) {
+		this.logger.error({
+			service: SERVICE_NAME,
+			method: "dispatch",
+			message: `Reactor ${reactor.name} failed for monitor ${evaluation.monitor.id}: ${error instanceof Error ? error.message : String(error)}`,
+			stack: error instanceof Error ? error.stack : undefined,
+		});
+	}
+
+	dispatch = async (evaluation: MonitorEvaluation) => {
+		for (const reactor of this.reactors) {
+			if (reactor.blocking) {
+				try {
+					await reactor.react(evaluation);
+				} catch (error: unknown) {
+					this.logFailure(reactor, evaluation, error);
+				}
+			} else {
+				reactor.react(evaluation).catch((error: unknown) => this.logFailure(reactor, evaluation, error));
+			}
+		}
+	};
+}

server/src/worker/reactors/reactor.incident.ts

@@ -0,0 +1,13 @@
+import { IMonitorReactor } from "@/worker/reactors/reactor.interface.js";
+import { MonitorEvaluation } from "@/worker/worker.interface.js";
+import { IIncidentService } from "@/domain/incidents/incident.service.js";
+
+export class IncidentReactor implements IMonitorReactor {
+	readonly name = "incident";
+	readonly blocking = true;
+	constructor(private incidentService: IIncidentService) {}
+
+	react = async (evaluation: MonitorEvaluation) => {
+		await this.incidentService.handleIncident(evaluation.monitor, evaluation.statusChange.code, evaluation.decision, evaluation.status);
+	};
+}

server/src/worker/reactors/reactor.interface.ts

@@ -0,0 +1,7 @@
+import { MonitorEvaluation } from "@/worker/worker.interface.js";
+
+export interface IMonitorReactor {
+	readonly name: string;
+	readonly blocking: boolean;
+	react(evaluation: MonitorEvaluation): Promise<void>;
+}

server/src/worker/reactors/reactor.notification.ts

@@ -0,0 +1,13 @@
+import { INotificationsService } from "@/domain/notifications/notification.service.js";
+import { IMonitorReactor } from "@/worker/reactors/reactor.interface.js";
+import { MonitorEvaluation } from "@/worker/worker.interface.js";
+export class NotificationReactor implements IMonitorReactor {
+	readonly name = "notification";
+	readonly blocking = false;
+	constructor(private notificationService: INotificationsService) {}
+
+	react = async (evaluation: MonitorEvaluation) => {
+		if (!evaluation.decision.shouldSendNotification) return;
+		await this.notificationService.handleNotifications(evaluation.monitor, evaluation.status, evaluation.decision);
+	};
+}

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

@@ -0,0 +1,187 @@
+import { Monitor } from "@/domain/monitors/monitor.types.js";
+import { MonitorEvaluation } from "@/worker/worker.interface.js";
+import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js";
+import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
+import { isWindowActive } from "@/utils/maintenanceWindow.js";
+import { ILogger } from "@/utils/logger.js";
+import { INetworkService } from "@/service/networkService.js";
+import { ICheckService } from "@/domain/checks/check.service.js";
+import { IBufferService } from "@/service/bufferService.js";
+import { IMonitorStatusPolicy } from "@/worker/worker.monitor-status-policy.js";
+import { IStatusService } from "@/service/statusService.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";
+
+const SERVICE_NAME = "CheckPipeline";
+
+export interface ICheckPipeline {
+	run(monitor: Monitor): Promise<MonitorEvaluation | null>; // null = skipped
+}
+
+export class CheckPipeline implements ICheckPipeline {
+	constructor(
+		private monitorsRepository: IMonitorsRepository,
+		private maintenanceWindowsRepository: IMaintenanceWindowsRepository,
+		private checkService: ICheckService,
+		private networkService: INetworkService,
+		private bufferService: IBufferService,
+		private monitorStatusPolicy: IMonitorStatusPolicy,
+		private statusService: IStatusService,
+		private logger: ILogger
+	) {}
+
+	private async isInMaintenanceWindow(monitorId: string, teamId: string) {
+		const maintenanceWindows = await this.maintenanceWindowsRepository.findByMonitorId(monitorId, teamId);
+		const now = new Date();
+		return maintenanceWindows.some((window) => isWindowActive(window, now));
+	}
+
+	run = async (monitor: Monitor): Promise<MonitorEvaluation | null> => {
+		const monitorId = monitor.id;
+		if (!monitorId) {
+			throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getMonitorJob" });
+		}
+		// ****************************
+		// Step 1:  Acquire
+		// ****************************
+
+		// Step 1a:  Maintenance window gate - skip if in maintenance
+		const maintenanceWindowActive = await this.isInMaintenanceWindow(monitor.id, monitor.teamId);
+		if (maintenanceWindowActive) {
+			this.logger.debug({
+				message: `Monitor ${monitor.id} is in maintenance window`,
+				service: SERVICE_NAME,
+				method: "runCheckPipeline",
+			});
+			if (monitor.status !== "maintenance") {
+				await this.monitorsRepository.updateById(monitor.id, monitor.teamId, { status: "maintenance" });
+			}
+			return null;
+		}
+
+		// Step 1b: Acquire status
+		const status = await this.networkService.requestStatus(monitor);
+		if (!status) {
+			throw new Error("No network response");
+		}
+
+		// ****************************
+		// Step 2: Record
+		// ****************************
+
+		// Step 2a:  Create & record a check, return null if fail
+		const check = this.checkService.buildCheck(status);
+		if (!check) {
+			this.logger.warn({
+				message: `No check could be built for monitor ${monitor.id}`,
+				service: SERVICE_NAME,
+				method: "runCHeckPipeline",
+				details: { code: status.code, message: status.message },
+			});
+			return null;
+		}
+		// Step 2b: Add to buffer
+		this.bufferService.addToBuffer(check);
+
+		// ****************************
+		// Step 3:  Evaluate and return result to reactors
+		// ****************************
+		const statusChangeResult = await this.statusService.updateMonitorStatus(status, check);
+
+		// Step 5.  Get decisions and create an evaluation obj
+		const decision = this.monitorStatusPolicy.evaluate(statusChangeResult);
+		const evaluation: MonitorEvaluation = {
+			monitor: statusChangeResult.monitor,
+			status,
+			check,
+			statusChange: statusChangeResult,
+			decision,
+		};
+		return evaluation;
+	};
+}
+
+export class GeoChecksPipeline implements ICheckPipeline {
+	constructor(
+		private maintenanceWindowsRepository: IMaintenanceWindowsRepository,
+		private geoChecksService: IGeoChecksService,
+		private bufferService: IBufferService,
+		private logger: ILogger
+	) {}
+
+	private async isInMaintenanceWindow(monitorId: string, teamId: string) {
+		const maintenanceWindows = await this.maintenanceWindowsRepository.findByMonitorId(monitorId, teamId);
+		const now = new Date();
+		return maintenanceWindows.some((window) => isWindowActive(window, now));
+	}
+
+	run = async (monitor: Monitor): Promise<MonitorEvaluation | null> => {
+		// ****************************
+		// Step 1:  Acquire
+		// ****************************
+
+		// Step 1a: Guards - skip if unsupported or not enabled
+
+		if (!monitor.id) {
+			throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getHeartbeatGeoJob" });
+		}
+
+		if (!monitor.geoCheckEnabled) {
+			return null;
+		}
+		if (!supportsGeoCheck(monitor.type)) {
+			this.logger.debug({
+				message: `Monitor ${monitor.id} type does not support geo checks, skipping`,
+				service: SERVICE_NAME,
+				method: "runGeoChecksPipeline",
+			});
+			return null;
+		}
+
+		if (!monitor.geoCheckLocations || monitor.geoCheckLocations.length === 0) {
+			this.logger.warn({
+				message: `No geo check locations configured for monitor ${monitor.id}`,
+				service: SERVICE_NAME,
+				method: "runGeoChecksPipeline",
+			});
+			return null;
+		}
+
+		// Step 1b: Maintenance window check
+		const maintenanceWindowActive = await this.isInMaintenanceWindow(monitor.id, monitor.teamId);
+		if (maintenanceWindowActive) {
+			this.logger.debug({
+				message: `Monitor ${monitor.id} is in maintenance window, skipping geo check`,
+				service: SERVICE_NAME,
+				method: "runGeoChecksPipeline",
+			});
+			return null;
+		}
+
+		// ****************************
+		// Step 2: Record
+		// ****************************
+
+		// Step 2a: Build geo check, return null if fail
+		const geoCheck = await this.geoChecksService.buildGeoCheck(monitor);
+		if (!geoCheck) {
+			this.logger.warn({
+				message: `No geo check could be built for monitor ${monitor.id}`,
+				service: SERVICE_NAME,
+				method: "runGeoChecksPipeline",
+			});
+			return null;
+		}
+
+		// Step 2b: Add  to buffer
+		this.bufferService.addGeoCheckToBuffer(geoCheck);
+
+		this.logger.debug({
+			message: `Geo check job executed for monitor ${monitor.id}`,
+			service: SERVICE_NAME,
+			method: "runGeoChecksPipeline",
+		});
+		return null;
+	};
+}

server/src/worker/worker.helper.ts

@@ -1,26 +1,17 @@
 const SERVICE_NAME = "JobQueueHelper";
 import type { Monitor } from "@/domain/monitors/monitor.types.js";
-import { supportsGeoCheck } from "@/domain/monitors/monitor.types.js";
-import { AppError } from "@/utils/AppError.js";
 import { ISettingsService } from "@/domain/app-settings/app-settings.service.js";
 import { ICheckService } from "@/domain/checks/check.service.js";
-import { type IGeoChecksService } from "@/domain/geo-checks/geo-check.service.js";
-import { IncidentService } from "@/domain/incidents/incident.service.js";
-import { INotificationsService } from "@/domain/notifications/notification.service.js";
-import { INetworkService } from "@/service/networkService.js";
-import { IStatusService } from "@/service/statusService.js";
 import { CHECK_TTL_SENTINEL } from "@/domain/checks/check.type.js";
-import { type StatusChangeResult } from "@/types/network.js";
-import { isWindowActive } from "@/utils/maintenanceWindow.js";
 import { IChecksRepository } from "@/domain/checks/check.repository.interface.js";
 import { IGeoChecksRepository } from "@/domain/geo-checks/geo-check.repository.interface.js";
 import { IIncidentsRepository } from "@/domain/incidents/incident.repository.interface.js";
-import { IMaintenanceWindowsRepository } from "@/domain/maintenance-windows/maintenance-window.repository.interface.js";
 import { IMonitorStatsRepository } from "@/domain/monitor-stats/monitor-stats.repository.interface.js";
 import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js";
 import { ITeamsRepository } from "@/domain/teams/team.repository.interface.js";
 import { ILogger } from "@/utils/logger.js";
-import { IBufferService } from "@/service/bufferService.js";
+import { IReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js";
+import { ICheckPipeline } from "@/worker/worker.check-pipeline.js";
 
 export interface IWorkerHelper {
 	readonly serviceName: string;
@@ -28,7 +19,6 @@ export interface IWorkerHelper {
 	getHeartbeatGeoJob(): (monitor: Monitor) => Promise<void>;
 	getCleanupOrphanedJob(): () => Promise<void>;
 	getCleanupRetentionJob(): () => Promise<void>;
-	isInMaintenanceWindow(monitorId: string, teamId: string): Promise<boolean>;
 }
 
 export interface MonitorActionDecision {
@@ -45,138 +35,59 @@ export interface MonitorActionDecision {
 	};
 }
 
-export class QueueHelper implements IWorkerHelper {
+export class WorkerHelper implements IWorkerHelper {
 	static SERVICE_NAME = SERVICE_NAME;
 
 	private logger: ILogger;
-	private networkService: INetworkService;
-	private statusService: IStatusService;
-	private notificationsService: INotificationsService;
 	private checkService: ICheckService;
 	private settingsService: ISettingsService;
-	private buffer: IBufferService;
-	private incidentService: IncidentService;
-	private maintenanceWindowsRepository: IMaintenanceWindowsRepository;
 	private monitorsRepository: IMonitorsRepository;
 	private teamsRepository: ITeamsRepository;
 	private monitorStatsRepository: IMonitorStatsRepository;
 	private checksRepository: IChecksRepository;
 	private incidentsRepository: IIncidentsRepository;
-	private geoChecksService: IGeoChecksService;
 	private geoChecksRepository: IGeoChecksRepository;
+	private reactorDispatcher: IReactorDispatcher;
+	private checkPipeline: ICheckPipeline;
+	private geoCheckPipeline: ICheckPipeline;
 
 	constructor(
 		logger: ILogger,
-		networkService: INetworkService,
-		statusService: IStatusService,
-		notificationsService: INotificationsService,
 		checkService: ICheckService,
 		settingsService: ISettingsService,
-		buffer: IBufferService,
-		incidentService: IncidentService,
-		maintenanceWindowsRepository: IMaintenanceWindowsRepository,
 		monitorsRepository: IMonitorsRepository,
 		teamsRepository: ITeamsRepository,
 		monitorStatsRepository: IMonitorStatsRepository,
 		checksRepository: IChecksRepository,
 		incidentsRepository: IIncidentsRepository,
-		geoChecksService: IGeoChecksService,
-		geoChecksRepository: IGeoChecksRepository
+		geoChecksRepository: IGeoChecksRepository,
+		reactorDispatcher: IReactorDispatcher,
+		checkPipeline: ICheckPipeline,
+		geoCheckPipeline: ICheckPipeline
 	) {
 		this.logger = logger;
-		this.networkService = networkService;
-		this.statusService = statusService;
 		this.checkService = checkService;
 		this.settingsService = settingsService;
-		this.buffer = buffer;
-		this.notificationsService = notificationsService;
-		this.incidentService = incidentService;
-		this.maintenanceWindowsRepository = maintenanceWindowsRepository;
 		this.monitorsRepository = monitorsRepository;
 		this.teamsRepository = teamsRepository;
 		this.monitorStatsRepository = monitorStatsRepository;
 		this.checksRepository = checksRepository;
 		this.incidentsRepository = incidentsRepository;
-		this.geoChecksService = geoChecksService;
 		this.geoChecksRepository = geoChecksRepository;
+		this.reactorDispatcher = reactorDispatcher;
+		this.checkPipeline = checkPipeline;
+		this.geoCheckPipeline = geoCheckPipeline;
 	}
 
 	get serviceName() {
-		return QueueHelper.SERVICE_NAME;
+		return WorkerHelper.SERVICE_NAME;
 	}
 
 	getHeartbeatJob = () => {
 		return async (monitor: Monitor) => {
 			try {
-				const monitorId = monitor.id;
-				const teamId = monitor.teamId;
-				if (!monitorId) {
-					throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getMonitorJob" });
-				}
-
-				// Step 1.  Check for maintenance window, if found, skip the check
-
-				const maintenanceWindowActive = await this.isInMaintenanceWindow(monitorId, teamId);
-				if (maintenanceWindowActive) {
-					this.logger.debug({
-						message: `Monitor ${monitorId} is in maintenance window`,
-						service: SERVICE_NAME,
-						method: "getMonitorJob",
-					});
-					if (monitor.status !== "maintenance") {
-						await this.monitorsRepository.updateById(monitorId, teamId, { status: "maintenance" });
-					}
-					return;
-				}
-
-				// Step 2.  Request monitor status
-				const status = await this.networkService.requestStatus(monitor);
-				if (!status) {
-					throw new Error("No network response");
-				}
-
-				// Step 3.  Build check
-				const check = this.checkService.buildCheck(status);
-				if (!check) {
-					this.logger.warn({
-						message: `No check could be built for monitor ${monitorId}`,
-						service: SERVICE_NAME,
-						method: "getMonitorJob",
-						details: { code: status.code, message: status.message },
-					});
-					return;
-				}
-				// Step 4 Add check to buffer
-				this.buffer.addToBuffer(check);
-				// Step 4.  Update monitor status
-				const statusChangeResult = await this.statusService.updateMonitorStatus(status, check);
-
-				// Step 5.  Get decisions
-				const decision = this.evaluateMonitorAction(statusChangeResult);
-
-				// Step 6. Handle notifications (best effort, continue even in event of failure, don't wait)
-				if (decision.shouldSendNotification) {
-					this.notificationsService.handleNotifications(statusChangeResult.monitor, status, decision).catch((error: unknown) => {
-						this.logger.error({
-							message: `Error sending notifications for job ${statusChangeResult.monitor.id}: ${error instanceof Error ? error.message : "Unknown error"}`,
-							service: SERVICE_NAME,
-							method: "getMonitorJob",
-							stack: error instanceof Error ? error.stack : undefined,
-						});
-					});
-				}
-
-				// Step 7. Handle incidents
-				try {
-					await this.incidentService.handleIncident(statusChangeResult.monitor, statusChangeResult.code, decision, status);
-				} catch (error: unknown) {
-					this.logger.warn({
-						message: `Error handling incident for job ${monitor.id}: ${error instanceof Error ? error.message : "Unknown error"}`,
-						service: SERVICE_NAME,
-						method: "getMonitorJob",
-						stack: error instanceof Error ? error.stack : undefined,
-					});
-				}
+				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",
@@ -189,6 +100,22 @@ export class QueueHelper implements IWorkerHelper {
 		};
 	};
 
+	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 {
@@ -280,86 +207,6 @@ export class QueueHelper implements IWorkerHelper {
 		};
 	};
 
-	getHeartbeatGeoJob = () => {
-		return async (monitor: Monitor) => {
-			try {
-				const monitorId = monitor.id;
-				const teamId = monitor.teamId;
-
-				// Step 1: Validate monitor eligibility
-				if (!monitorId) {
-					throw new AppError({ message: "No monitor id", service: SERVICE_NAME, method: "getHeartbeatGeoJob" });
-				}
-
-				if (!monitor.geoCheckEnabled) {
-					return;
-				}
-				if (!supportsGeoCheck(monitor.type)) {
-					this.logger.debug({
-						message: `Monitor ${monitorId} type does not support geo checks, skipping`,
-						service: SERVICE_NAME,
-						method: "getHeartbeatGeoJob",
-					});
-					return;
-				}
-
-				if (!monitor.geoCheckLocations || monitor.geoCheckLocations.length === 0) {
-					this.logger.warn({
-						message: `No geo check locations configured for monitor ${monitorId}`,
-						service: SERVICE_NAME,
-						method: "getHeartbeatGeoJob",
-					});
-					return;
-				}
-
-				// Step 2: Check for maintenance window
-				const maintenanceWindowActive = await this.isInMaintenanceWindow(monitorId, teamId);
-				if (maintenanceWindowActive) {
-					this.logger.debug({
-						message: `Monitor ${monitorId} is in maintenance window, skipping geo check`,
-						service: SERVICE_NAME,
-						method: "getHeartbeatGeoJob",
-					});
-					return;
-				}
-
-				// Step 3: Build geo check (handles API calls and polling)
-				const geoCheck = await this.geoChecksService.buildGeoCheck(monitor);
-				if (!geoCheck) {
-					this.logger.warn({
-						message: `No geo check could be built for monitor ${monitorId}`,
-						service: SERVICE_NAME,
-						method: "getHeartbeatGeoJob",
-					});
-					return;
-				}
-
-				// Step 4: Add geo check to buffer
-				this.buffer.addGeoCheckToBuffer(geoCheck);
-
-				this.logger.debug({
-					message: `Geo check job executed for monitor ${monitorId}`,
-					service: SERVICE_NAME,
-					method: "getHeartbeatGeoJob",
-				});
-			} 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
-			}
-		};
-	};
-
-	async isInMaintenanceWindow(monitorId: string, teamId: string) {
-		const maintenanceWindows = await this.maintenanceWindowsRepository.findByMonitorId(monitorId, teamId);
-		const now = new Date();
-		return maintenanceWindows.some((window) => isWindowActive(window, now));
-	}
-
 	getCleanupRetentionJob = () => {
 		return async () => {
 			try {
@@ -393,42 +240,4 @@ export class QueueHelper implements IWorkerHelper {
 			}
 		};
 	};
-
-	private evaluateMonitorAction(statusChangeResult: StatusChangeResult): MonitorActionDecision {
-		const { monitor, statusChanged, prevStatus } = statusChangeResult;
-
-		// Initialize result
-		const decision: MonitorActionDecision = {
-			shouldCreateIncident: false,
-			shouldResolveIncident: false,
-			shouldSendNotification: false,
-			incidentReason: null,
-			notificationReason: null,
-		};
-
-		if (!statusChanged) {
-			return decision;
-		}
-
-		if (monitor.status === "down") {
-			// Monitor went down (unreachable)
-			decision.shouldCreateIncident = true;
-			decision.shouldSendNotification = true;
-			decision.incidentReason = "status_down";
-			decision.notificationReason = "status_change";
-		} else if (monitor.status === "breached") {
-			// Hardware monitor exceeded thresholds
-			decision.shouldCreateIncident = true;
-			decision.shouldSendNotification = true;
-			decision.incidentReason = "threshold_breach";
-			decision.notificationReason = "threshold_breach";
-		} else if (monitor.status === "up" && (prevStatus === "down" || prevStatus === "breached")) {
-			// Monitor recovered from down or breached state
-			decision.shouldResolveIncident = true;
-			decision.shouldSendNotification = true;
-			decision.notificationReason = "status_change";
-		}
-
-		return decision;
-	}
 }

server/src/worker/worker.interface.ts

@@ -1,5 +1,16 @@
 import { Monitor } from "@/domain/monitors/monitor.types.js";
 import { QueueWorker } from "@/domain/queue-workers/queue-worker.type.js";
+import { Check } from "@/domain/checks/check.type.js";
+import { MonitorPayloadMap, MonitorStatusResponse, StatusChangeResult } from "@/types/network.js";
+import { MonitorActionDecision } from "@/worker/worker.helper.js";
+
+export type MonitorEvaluation = {
+	monitor: Monitor;
+	status: MonitorStatusResponse<MonitorPayloadMap[keyof MonitorPayloadMap]>; // raw result from networkService.requestStatus
+	check: Check;
+	statusChange: StatusChangeResult; // from statusService.updateMonitorStatus
+	decision: MonitorActionDecision; // from MonitorStatusPolicy
+};
 
 export type WorkerJobFailure = {
 	monitorId: string | number;

server/src/worker/worker.monitor-status-policy.ts

@@ -0,0 +1,46 @@
+import { StatusChangeResult } from "@/types/network.js";
+import { MonitorActionDecision } from "@/worker/worker.helper.js";
+
+export interface IMonitorStatusPolicy {
+	evaluate(statusChange: StatusChangeResult): MonitorActionDecision;
+}
+
+export class MonitorStatusPolicy implements IMonitorStatusPolicy {
+	evaluate = (statusChangeResult: StatusChangeResult): MonitorActionDecision => {
+		const { monitor, statusChanged, prevStatus } = statusChangeResult;
+
+		// Initialize result
+		const decision: MonitorActionDecision = {
+			shouldCreateIncident: false,
+			shouldResolveIncident: false,
+			shouldSendNotification: false,
+			incidentReason: null,
+			notificationReason: null,
+		};
+
+		if (!statusChanged) {
+			return decision;
+		}
+
+		if (monitor.status === "down") {
+			// Monitor went down (unreachable)
+			decision.shouldCreateIncident = true;
+			decision.shouldSendNotification = true;
+			decision.incidentReason = "status_down";
+			decision.notificationReason = "status_change";
+		} else if (monitor.status === "breached") {
+			// Hardware monitor exceeded thresholds
+			decision.shouldCreateIncident = true;
+			decision.shouldSendNotification = true;
+			decision.incidentReason = "threshold_breach";
+			decision.notificationReason = "threshold_breach";
+		} else if (monitor.status === "up" && (prevStatus === "down" || prevStatus === "breached")) {
+			// Monitor recovered from down or breached state
+			decision.shouldResolveIncident = true;
+			decision.shouldSendNotification = true;
+			decision.notificationReason = "status_change";
+		}
+
+		return decision;
+	};
+}