import { Injectable, OnModuleInit, OnModuleDestroy, Logger } from '@nestjs/common'; import { RedisService } from '../infrastructure/redis/redis.service'; import { QUEUE_DEFINITIONS } from '../infrastructure/queue/queue-definitions'; import * as os from 'os'; export interface WorkerHeartbeatData { instanceId: string; appVersion: string; startedAt: string; queues: string[]; concurrency: number; activeJobCount: number; hostname: string; } export interface WorkerStatus { instanceId: string; appVersion: string; startedAt: string; queues: string[]; hostname: string; lastHeartbeatAgo: number; // seconds since last beat status: 'online' | 'stale' | 'offline'; } @Injectable() export class WorkerHeartbeatService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(WorkerHeartbeatService.name); private intervalId: ReturnType | null = null; private startedAt: Date; private static readonly HEARTBEAT_INTERVAL_MS = 30_000; private static readonly HEARTBEAT_TTL_SEC = 90; constructor(private readonly redis: RedisService) {} async onModuleInit() { this.startedAt = new Date(); await this.beat(); this.intervalId = setInterval(() => this.beat(), WorkerHeartbeatService.HEARTBEAT_INTERVAL_MS); this.logger.log(`Heartbeat started — instance=${this.resolveInstanceId()}`); } async onModuleDestroy() { if (this.intervalId) { clearInterval(this.intervalId); this.intervalId = null; } try { await this.redis.del(this.heartbeatKey()); } catch { // Best-effort cleanup } this.logger.log('Heartbeat stopped'); } // ── Public static helpers for Admin API ── static heartbeatKeyPrefix = 'worker:heartbeat:'; static heartbeatKeyFor(instanceId: string): string { return `${WorkerHeartbeatService.heartbeatKeyPrefix}${instanceId}`; } static async computeStatuses(redis: RedisService): Promise { const keys = await redis.keys(`${WorkerHeartbeatService.heartbeatKeyPrefix}*`); const now = Date.now(); const statuses: WorkerStatus[] = []; for (const key of keys) { const raw = await redis.get(key); if (!raw) continue; let data: WorkerHeartbeatData; try { data = JSON.parse(raw); } catch { continue; } const ttl = await redis.ttl(key); const lastHeartbeatAgo = WorkerHeartbeatService.HEARTBEAT_TTL_SEC - ttl; let status: WorkerStatus['status']; if (ttl < 0) { status = 'offline'; } else if (ttl <= 30) { // TTL ≤ 30s means haven't heard from worker in 60+ seconds status = 'stale'; } else { status = 'online'; } statuses.push({ instanceId: data.instanceId, appVersion: data.appVersion, startedAt: data.startedAt, queues: data.queues, hostname: data.hostname, lastHeartbeatAgo, status, }); } return statuses; } // ── Private ── private resolveInstanceId(): string { return process.env.WORKER_INSTANCE_ID || `worker-${os.hostname()}-${process.pid}`; } private heartbeatKey(): string { return WorkerHeartbeatService.heartbeatKeyFor(this.resolveInstanceId()); } private async beat() { const instanceId = this.resolveInstanceId(); const data: WorkerHeartbeatData = { instanceId, // eslint-disable-next-line @typescript-eslint/no-var-requires appVersion: (() => { try { return require('../../../package.json').version; } catch { return '0.0.0'; } })(), startedAt: this.startedAt.toISOString(), queues: Object.keys(QUEUE_DEFINITIONS), concurrency: Object.values(QUEUE_DEFINITIONS)[0]?.workerOptions.concurrency ?? 1, activeJobCount: 0, hostname: os.hostname(), }; try { await this.redis.set(this.heartbeatKey(), JSON.stringify(data), WorkerHeartbeatService.HEARTBEAT_TTL_SEC); } catch (err: any) { this.logger.warn(`Heartbeat write failed: ${err.message}`); } } }