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

import { supportsGeoCheck } from "@/domain/monitors/monitor.types.js";
import { IQueueWorker } from "@/worker/worker.interface.js";
import { type Job, type JobSeed, type JobType, jobId, LOCK_MS } from "@/domain/jobs/job.type.js";
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";
import { IQueueWorkersRepository } from "@/domain/queue-workers/queue-worker.repository.interface.js";
import { WORKER_TTL_SECONDS } from "@/domain/queue-workers/queue-worker.model.js";
import { QueueMode } from "@/domain/app-settings/app-settings.type.js";
import { JobScheduler } from "@/worker/worker.job-scheduler.js";
const SERVICE_NAME = "JobQueue";
const POLL_MS = 250; // base poll interval while a loop is actively claiming work
const POLL_MAX_MS = 5000; // back off poll interval to 5s if no jobs
const WORKER_STALE_MS = WORKER_TTL_SECONDS * 1000; // a worker counts as alive if seen within this window
const HEARTBEAT_MS = WORKER_STALE_MS / 3; // Worker can miss two beats without being considered stale
const LOCK_RENEW_MS = LOCK_MS / 3; // renew an in-flight job's lock at 1/3 the lease, so it survives two missed renewals
const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000;
const CONCURRENCY: Record<JobType, number> = {
	check: 500,
	"geo-check": 20,
	evaluate: 20,
	"cleanup-orphaned": 1,
	"cleanup-retention": 1,
};

export class DBQueueWorker extends JobScheduler implements IQueueWorker {
	static SERVICE_NAME = SERVICE_NAME;
	private inFlight: Record<JobType, number> = {
		check: 0,
		"geo-check": 0,
		evaluate: 0,
		"cleanup-orphaned": 0,
		"cleanup-retention": 0,
	};

	constructor(
		private logger: ILogger,
		jobsRepository: IJobsRepository, // ← no modifier: forwarded to super
		private monitorsRepository: IMonitorsRepository,
		private checksRepository: IChecksRepository,
		private checkService: ICheckService,
		private checkProducer: ICheckProducer,
		private checkEvaluator: ICheckEvaluator,
		private geoCheckPipeline: ICheckPipeline,
		private dispatcher: IReactorDispatcher,
		private helper: IWorkerHelper,
		queueWorkersRepository: IQueueWorkersRepository, // ← no modifier: forwarded
		private queueMode: QueueMode,
		private queuePrimaryProcesses: boolean,
		workerId: string
	) {
		super(jobsRepository, queueWorkersRepository, workerId);
	}

	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,
		queueWorkersRepository: IQueueWorkersRepository,
		queueMode: QueueMode,
		queuePrimaryProcesses: boolean,
		workerId: string
	): Promise<DBQueueWorker> {
		const instance = new DBQueueWorker(
			logger,
			jobsRepository,
			monitorsRepository,
			checksRepository,
			checkService,
			checkProducer,
			checkEvaluator,
			geoCheckPipeline,
			dispatcher,
			helper,
			queueWorkersRepository,
			queueMode,
			queuePrimaryProcesses,
			workerId
		);
		await instance.init();
		return instance;
	}

	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,
	});

	// ********************
	// 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.findByIdLean(job.refId); // job row has no teamId
		if (!monitor) return;
		await this.checkProducer.produce(monitor);
	};

	// ********************
	// Stage 2:  Evaluator loop stays here
	// ********************

	private runEvaluate = async (job: Job) => {
		if (!job.refId) return;
		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, current);
			await this.dispatcher.dispatch(evaluation); // Handle incidents and notifications
			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.findByIdLean(job.refId!);
		if (monitor) await this.geoCheckPipeline.run(monitor); // returns null; no evaluate handoff
	};

	// Extend the lock while a job runs so a slow-but-alive job is never reclaimed.
	private renewLock = async (job: Job): Promise<boolean> => {
		try {
			const held = await this.jobsRepository.renewLocks([job.id], Date.now());
			if (held === 0) {
				this.logger.warn({ message: `Lost lock on job ${job.id}, stopping renewal`, service: SERVICE_NAME, method: "renewLock" });
				return false;
			}
			return true;
		} catch (error: unknown) {
			this.logger.warn({ message: error instanceof Error ? error.message : String(error), service: SERVICE_NAME, method: "renewLock" });
			return true;
		}
	};

	private runJob = async (job: Job) => {
		const renewTimer = setInterval(async () => {
			const renewLock = await this.renewLock(job);
			if (!renewLock) clearInterval(renewTimer);
		}, LOCK_RENEW_MS);
		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}`,
			});
		} finally {
			clearInterval(renewTimer);
		}
	};

	private startLoop = (type: JobType) => {
		this.pollMs[type] = POLL_MS;
		const tick = async () => {
			if (this.stopped) return;
			this.ticking[type] = true;
			try {
				// Claim a batch sized to the free capacity
				const capacity = CONCURRENCY[type] - this.inFlight[type];
				const jobs = await this.jobsRepository.claimDueBatch(type, capacity, Date.now());
				for (const job of jobs) {
					this.inFlight[type]++; // stay within concurrency limits
					this.runJob(job).finally(() => {
						this.inFlight[type]--; // job done, free the slot
					});
				}
				// Back off if no jobs found, reset to base if jobs found
				this.pollMs[type] = capacity > 0 && jobs.length === 0 ? Math.min(this.pollMs[type] * 2, POLL_MAX_MS) : POLL_MS;
			} catch (error: unknown) {
				this.logger.error({
					message: error instanceof Error ? error.message : String(error),
					service: SERVICE_NAME,
					method: `loop:${type}`,
				});
			}
			// Clearing the flag and re-arming run synchronously (no await between), so wake() can never interleave here.
			this.ticking[type] = false;
			if (!this.stopped) this.timers.set(type, setTimeout(tick, this.pollMs[type])); // re-arm unless shutting down
		};
		this.tickFns.set(type, tick);
		tick();
	};

	// Wake a worker loop after adding a job that's due immediately.
	// Otherwise the loop could be delayed by POLL_MAX_MS if it is already fully backed off

	// 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));
			}
		}
		// 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
	private heartbeat = async () => {
		try {
			await this.queueWorkersRepository.upsert(this.workerId, this.queueMode);
		} catch (error: unknown) {
			this.logger.warn({
				message: error instanceof Error ? error.message : String(error),
				service: SERVICE_NAME,
				method: "heartbeat",
			});
		}
	};

	init = async () => {
		this.stopped = false;

		await this.heartbeat();
		this.timers.set("heartbeat", setInterval(this.heartbeat, HEARTBEAT_MS));

		if (this.queueMode === "primary") {
			await this.reconcile();
		}

		if (this.queuePrimaryProcesses === true || this.queueMode === "worker") {
			for (const type of Object.keys(this.inFlight) as JobType[]) {
				this.startLoop(type);
			}
		}
		return true;
	};

	override flushQueues = async () => {
		await this.shutdown();
		const ok = await this.init();
		return { success: ok };
	};
}