add health server and metrics to db queue worker

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

@@ -36,6 +36,10 @@ const CONCURRENCY: Record<JobType, number> = {
 
 export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 	static SERVICE_NAME = SERVICE_NAME;
+
+	private initComplete: boolean = false;
+	private lastTickAt: number | null = null;
+
 	private inFlight: Record<JobType, number> = {
 		check: 0,
 		"geo-check": 0,
@@ -46,6 +50,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 
 	constructor(
 		private logger: ILogger,
+		private isDbConnected: () => boolean,
 		jobsRepository: IJobsRepository, // ← no modifier: forwarded to super
 		private monitorsRepository: IMonitorsRepository,
 		private checksRepository: IChecksRepository,
@@ -70,6 +75,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 
 	static async create(
 		logger: ILogger,
+		isDbConnected: () => boolean,
 		jobsRepository: IJobsRepository,
 		monitorsRepository: IMonitorsRepository,
 		checksRepository: IChecksRepository,
@@ -87,6 +93,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 	): Promise<DBQueueWorker> {
 		const instance = new DBQueueWorker(
 			logger,
+			isDbConnected,
 			jobsRepository,
 			monitorsRepository,
 			checksRepository,
@@ -213,6 +220,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 	private startLoop = (type: JobType) => {
 		this.pollMs[type] = POLL_MS;
 		const tick = async () => {
+			this.lastTickAt = Date.now();
 			if (this.stopped) return;
 			this.ticking[type] = true;
 			try {
@@ -288,6 +296,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 				this.startLoop(type);
 			}
 		}
+		this.initComplete = true;
 		return true;
 	};
 
@@ -298,6 +307,7 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 	};
 
 	override drain = async () => {
+		this.draining = true;
 		this.stopped = true;
 		const cutoff = Date.now() + DRAIN_TIMEOUT_MS;
 		while (this.getInFlightCount() > 0 && Date.now() < cutoff) {
@@ -326,11 +336,11 @@ export class DBQueueWorker extends JobScheduler implements IQueueWorker {
 		return {
 			workerId: this.workerId,
 			mode: this.queueMode,
-			dbConnected: true, // TODO
-			initComplete: true, // TODO
-			draining: false, // TODO
-			lastTickAt: null, // TODO
-			inFlight: Number.MAX_SAFE_INTEGER, // TODO
+			dbConnected: this.isDbConnected(),
+			initComplete: this.initComplete,
+			draining: this.draining,
+			lastTickAt: this.lastTickAt,
+			inFlight: this.getInFlightCount(),
 		};
 	};
 }

server/src/worker/worker.health-server.ts

@@ -0,0 +1,74 @@
+import http from "http";
+import { IQueueWorker, WorkerHealth } from "@/worker/worker.interface.js";
+import { ILogger } from "@/utils/logger.js";
+
+const SERVICE_NAME = "HealthServer";
+const LIVENESS_STALE_MS = 30_000; // 30s ~6x POLL_MAX_MS so we don't restart workers
+
+export interface IHealthServer {
+	listen(): Promise<void>;
+	close(): Promise<void>;
+}
+
+const isLive = (health: WorkerHealth): boolean => {
+	if (!health.initComplete) return false; // Not init yet
+	if (health.draining) return true; // Draining workers are still healthy
+
+	if (health.lastTickAt === null) return false; // Never ticked
+
+	if (Date.now() - health.lastTickAt >= LIVENESS_STALE_MS) return false; // Stale if last tick exceed LIVENESS_STALE_MS
+
+	return true;
+};
+
+const isReady = (health: WorkerHealth): boolean => {
+	if (health.draining) return false; // Draining workers are not ready
+	return health.initComplete && health.dbConnected;
+};
+
+export class HealthServer implements IHealthServer {
+	private server: http.Server;
+
+	constructor(
+		private logger: ILogger,
+		private port: string,
+		private worker: IQueueWorker
+	) {
+		this.server = http.createServer(this.handle);
+	}
+
+	private handle = (req: http.IncomingMessage, res: http.ServerResponse) => {
+		const path = (req.url ?? "").split("?")[0]; // Strip query params
+		if (path !== "/livez" && path !== "/readyz") {
+			res.writeHead(404).end();
+			return;
+		}
+
+		// Get health
+		const health = this.worker.getHealth();
+		const ok = path === "/livez" ? isLive(health) : isReady(health);
+		const code = ok ? 200 : 503;
+		res.writeHead(code, { "Content-Type": "application/json" });
+		res.end(JSON.stringify(health));
+	};
+
+	listen = () => {
+		return new Promise<void>((resolve) => {
+			const port = Number(this.port) || 52346;
+			this.server.listen(port, () => {
+				this.logger.info({
+					message: `Health server listening on ${port}`,
+					service: SERVICE_NAME,
+					method: "listen",
+				});
+				resolve();
+			});
+		});
+	};
+
+	close = () => {
+		return new Promise<void>((resolve) => {
+			this.server.close(() => resolve());
+		});
+	};
+}

server/src/worker/worker.job-scheduler.ts

@@ -23,6 +23,7 @@ export class JobScheduler implements IJobScheduler {
 	}
 
 	protected stopped = false;
+	protected draining = false;
 	protected ticking: Partial<Record<JobType, boolean>> = {}; // Whether a loop is mid tick
 	protected tickFns = new Map<JobType, () => void>(); // Store tick fns so we can run them immediately on wake
 	protected timers = new Map<JobType | "heartbeat", NodeJS.Timeout>();
@@ -156,6 +157,7 @@ export class JobScheduler implements IJobScheduler {
 
 	drain = async () => {
 		this.stopped = true;
+		this.draining = true;
 	};
 
 	shutdown = async () => {