diff --git a/src/index.ts b/src/index.ts index 7a93880d..31d16142 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,5 +1,6 @@ import cluster from 'node:cluster' import { EventEmitter } from 'node:events' +import fs from 'node:fs' import http from 'node:http' import { resolve as resolvePath } from 'node:path' import { pathToFileURL } from 'node:url' @@ -294,9 +295,14 @@ const getTrackCacheManagerClass = async (): Promise< let config: NodelinkConfig -const loadConfig = async (): Promise => { - const resolveRootConfigUrl = (fileName: string): string => - pathToFileURL(resolvePath(process.cwd(), fileName)).href +const loadConfig = async (updateTimestamp?: number): Promise => { + const resolveRootConfigUrl = (fileName: string): string => { + let url = pathToFileURL(resolvePath(process.cwd(), fileName)).href + if (updateTimestamp) { + url += `?update=${updateTimestamp}` + } + return url + } const resolveConfigExport = ( importedModule: Record, fileName: string @@ -424,6 +430,68 @@ config = await loadConfig() // Apply environment variable overrides after config is loaded applyEnvOverrides(config as unknown as Record) +let watcherTimeout: NodeJS.Timeout | null = null + +const setupConfigWatcher = (): void => { + const configPath = process.cwd() + logger('info', 'Server', 'Initializing configuration file watcher...') + + try { + fs.watch(configPath, { recursive: false }, (eventType, filename) => { + if (!filename || (filename !== 'config.ts' && filename !== 'config.js')) return + + if (watcherTimeout) clearTimeout(watcherTimeout) + + watcherTimeout = setTimeout(async () => { + logger('info', 'Server', `Configuration file change detected: ${filename}. Reloading...`) + try { + const timestamp = Date.now() + const newConfig = await loadConfig(timestamp) + applyEnvOverrides(newConfig as unknown as Record) + + initLogger(newConfig as unknown as { logging?: import('./typings/utils.types.ts').LoggingConfig }) + + config = newConfig + + const nodelinkInstance = (global as typeof globalThis & { nodelink?: NodelinkServer }).nodelink + if (nodelinkInstance) { + nodelinkInstance.options = newConfig + + if (nodelinkInstance.sources) { + await nodelinkInstance.sources.loadFolder() + } + if (nodelinkInstance.lyrics) { + await nodelinkInstance.lyrics.loadFolder() + } + if (nodelinkInstance.meanings) { + await nodelinkInstance.meanings.loadFolder() + } + + if (nodelinkInstance.workerManager) { + // @ts-ignore + nodelinkInstance.workerManager.updateConfig(newConfig) + } + + logger('info', 'Server', 'Configuration hot-reloaded successfully.') + } + } catch (err: unknown) { + logger( + 'error', + 'Server', + `Failed to reload configuration: ${err instanceof Error ? err.message : String(err)}` + ) + } + }, 200) + }) + } catch (watchError: unknown) { + logger( + 'error', + 'Server', + `Failed to initialize config watcher: ${watchError instanceof Error ? watchError.message : String(watchError)}` + ) + } +} + const clusterEnabled = // biome-ignore lint/complexity/useLiteralKeys: TypeScript requires index signature access process.env['CLUSTER_ENABLED']?.toLowerCase() === 'true' || @@ -2416,6 +2484,7 @@ if (clusterEnabled && cluster.isPrimary) { await nserver.start({ isClusterPrimary: true }) ;(global as typeof globalThis & { nodelink?: NodelinkServer }).nodelink = nserver + setupConfigWatcher() let isShuttingDown = false const shutdown = async () => { @@ -2490,6 +2559,7 @@ if (clusterEnabled && cluster.isPrimary) { await nserver.start() ;(global as typeof globalThis & { nodelink?: NodelinkServer }).nodelink = nserver + setupConfigWatcher() logger( 'info', diff --git a/src/managers/workerManager.ts b/src/managers/workerManager.ts index e285ea57..04b16c90 100644 --- a/src/managers/workerManager.ts +++ b/src/managers/workerManager.ts @@ -1794,6 +1794,55 @@ export default class WorkerManager { return workerMetrics } + updateConfig(config: NodelinkConfig): void { + this.config = config + const availableParallelism = + typeof os.availableParallelism === 'function' + ? os.availableParallelism() + : os.cpus().length + this.maxWorkers = + config.cluster.workers === 0 + ? availableParallelism + : Math.max(1, config.cluster.workers || 0) + this.minWorkers = Math.max(1, config.cluster?.minWorkers || 1) + this.commandTimeout = config.cluster?.commandTimeout || 45000 + this.fastCommandTimeout = config.cluster?.fastCommandTimeout || 10000 + this.maxRetries = config.cluster?.maxRetries || 2 + this.scalingConfig = { + maxPlayersPerWorker: + config.cluster.scaling?.maxPlayersPerWorker || + config.cluster.workers || + 20, + targetUtilization: config.cluster.scaling?.targetUtilization || 0.7, + scaleUpThreshold: config.cluster.scaling?.scaleUpThreshold || 0.75, + scaleDownThreshold: config.cluster.scaling?.scaleDownThreshold || 0.3, + idleWorkerTimeoutMs: config.cluster.scaling?.idleWorkerTimeoutMs || 60000, + checkIntervalMs: config.cluster.scaling?.checkIntervalMs || 5000, + lagPenaltyLimit: config.cluster.scaling?.lagPenaltyLimit || 60, + cpuPenaltyLimit: config.cluster.scaling?.cpuPenaltyLimit || 0.85 + } + + logger( + 'info', + 'Cluster', + `WorkerManager config updated. Min: ${this.minWorkers}, Max: ${this.maxWorkers} workers.` + ) + + for (const worker of this.workers) { + if (worker && typeof worker.send === 'function') { + try { + worker.send({ type: 'updateConfig', payload: config }) + } catch (e: unknown) { + logger( + 'error', + 'Cluster', + `Failed to send updateConfig to worker ${worker.id}: ${e instanceof Error ? e.message : String(e)}` + ) + } + } + } + } + destroy(): void { this.isDestroying = true this._stopScalingCheck() diff --git a/src/workers/main.ts b/src/workers/main.ts index c6454775..43266384 100644 --- a/src/workers/main.ts +++ b/src/workers/main.ts @@ -2499,6 +2499,31 @@ process.on('message', (msg: unknown) => { ) return } + if (message.type === 'updateConfig') { + const newConfig = message.payload as unknown as NodeLinkConfig + if (newConfig) { + config = newConfig + nodelink.options = newConfig as any + + applyEnvOverrides(config as unknown as Record) + + const logging = config.logging as LoggingConfig + initLogger({ logging }) + + nodelink.sources?.loadFolder?.().catch((err: any) => { + logger('error', 'Worker', `Failed to reload sources folder: ${err.message}`) + }) + nodelink.lyrics?.loadFolder?.().catch((err: any) => { + logger('error', 'Worker', `Failed to reload lyrics folder: ${err.message}`) + }) + nodelink.meanings?.loadFolder?.().catch((err: any) => { + logger('error', 'Worker', `Failed to reload meanings folder: ${err.message}`) + }) + + logger('info', 'Worker', 'Configuration dynamically updated from Master.') + } + return + } if (message.type === 'ping') { if (process.connected) { try {