import { processDestructions, processCreations } from '../fleet/create.js'; import { sweep } from './sweep.js'; import { betaTick } from './beta-runner.js'; import { emit } from './bus.js'; import type { ServerStatusEntry } from './bus.js'; import { rconStatusResult } from '../integrations/rcon.js'; import type { RconFailure } from '../integrations/rcon.js'; import { listServers, setServerField } from '../fleet/servers.js'; import { komodoInspectHealthy } from '../integrations/komodo.js'; import { restartSettling } from './restart-rollout.js'; import { classifyHealth } from '../fleet/health.js'; import { STATUS_REFRESH_MS } from '../../server-freshness.js'; const TICK_MS = 15_000; // Cap concurrent UDP RCON calls so a 15-server fleet doesn't blast the // host all at once. const STATUS_CONCURRENCY = 4; let started = false; let tickTimer: NodeJS.Timeout | null = null; let statusTimer: NodeJS.Timeout | null = null; let tickInFlight: Promise | null = null; function tick(): Promise { if (tickInFlight) return tickInFlight; tickInFlight = runTick().finally(() => { tickInFlight = null; }); return tickInFlight; } async function runTick(): Promise { try { await processCreations(); } catch (err) { console.error('[worker] creation error:', err); } try { await processDestructions(); } catch (err) { console.error('[worker] destruction error:', err); } try { await sweep(); } catch (err) { console.error('[worker] sweep error:', err); } try { await betaTick(); } catch (err) { console.error('[worker] beta-runner error:', err); } } /** * Encode a structured RCON failure as `"reason:detail"` so `last_error` * stays a single column but the UI can split it apart when rendering the * diagnostic panel. Reasons are stable; details are human-facing. */ function encodeError(reason: RconFailure, detail: string): string { return `${reason}:${detail}`; } let statusInFlight: Promise | null = null; export function fanoutStatus(): Promise { if (statusInFlight) return statusInFlight; statusInFlight = refreshStatus().finally(() => { statusInFlight = null; }); return statusInFlight; } async function refreshStatus(): Promise { const servers = listServers(); if (servers.length === 0) return; const entries: ServerStatusEntry[] = new Array(servers.length); let cursor = 0; async function pump(): Promise { while (true) { const i = cursor++; if (i >= servers.length) return; const srv = servers[i]; const container = restartSettling(srv.name) ? { ok: false as const, state: 'restarting', reason: 'Restart requested' } : await komodoInspectHealthy(srv.name).catch((err) => ({ ok: false as const, reason: (err as Error).message })); const containerState = container.ok ? 'running' : ('state' in container && typeof container.state === 'string' ? container.state.toLowerCase() : 'unknown'); setServerField(srv.name, 'container_state', containerState); if (!container.ok) { // Container readiness precedes RCON readiness. Old replies and // failures describe the previous process, not this startup. setServerField(srv.name, 'last_error', null); setServerField(srv.name, 'last_reply_snippet', null); entries[i] = { server: srv.name, players: null, map: srv.map, gametype: null, roster: null, health: classifyHealth(null, null, containerState), lastSeenAt: srv.last_seen_at, lastError: null, endpoint: null, reason: null, replySnippet: null, activeSince: null, inactiveSince: null, gameStartedAt: null, gameEndedAt: null }; continue; } const result = await rconStatusResult(srv.name).catch((err) => ({ ok: false as const, endpoint: null, reason: 'send-error' as RconFailure, detail: (err as Error).message, rawReply: undefined as string | undefined })); const now = new Date().toISOString(); if (result.ok) { // Compare against the previously persisted count BEFORE we // overwrite last_players — occupancy transitions are anchored // to that boundary. const prevPlayers = srv.last_players; const nextPlayers = result.status.players; let activeSince = srv.active_since; let inactiveSince = srv.inactive_since; const gameStartedAt = srv.game_started_at; const gameEndedAt = srv.game_ended_at; if (nextPlayers > 0 && (activeSince === null || prevPlayers === 0 || prevPlayers === null)) { activeSince = now; inactiveSince = null; setServerField(srv.name, 'active_since', activeSince); setServerField(srv.name, 'inactive_since', null); } else if (nextPlayers === 0 && (inactiveSince === null || (prevPlayers !== null && prevPlayers > 0))) { inactiveSince = now; activeSince = null; setServerField(srv.name, 'inactive_since', inactiveSince); setServerField(srv.name, 'active_since', null); } // Persist last-observed state. last_seen_at is the health anchor. setServerField(srv.name, 'last_players', nextPlayers); setServerField(srv.name, 'last_map', result.status.map); setServerField(srv.name, 'last_gametype', result.status.gametype); setServerField(srv.name, 'last_roster', JSON.stringify(result.status.roster)); setServerField(srv.name, 'last_seen_at', now); setServerField(srv.name, 'last_checked_at', now); setServerField(srv.name, 'last_error', null); setServerField(srv.name, 'last_reply_snippet', result.rawReply); entries[i] = { server: srv.name, players: nextPlayers, map: result.status.map, gametype: result.status.gametype, roster: result.status.roster, health: classifyHealth(now, null), lastSeenAt: now, lastError: null, endpoint: result.endpoint, reason: null, replySnippet: result.rawReply, activeSince, inactiveSince, gameStartedAt, gameEndedAt }; } else { // RCON unreachable — leave last_seen_at, active_since, and // inactive_since alone (health decays to offline naturally; // occupancy timers survive transient outages). Record the // structured failure. const encoded = encodeError(result.reason, result.detail); const snippet = result.rawReply ?? null; setServerField(srv.name, 'last_checked_at', now); setServerField(srv.name, 'last_error', encoded); setServerField(srv.name, 'last_reply_snippet', snippet); entries[i] = { server: srv.name, players: srv.last_players, map: srv.last_map, gametype: srv.last_gametype, roster: srv.last_roster ? safeJson(srv.last_roster) : null, health: classifyHealth(srv.last_seen_at, encoded), lastSeenAt: srv.last_seen_at, lastError: encoded, endpoint: result.endpoint, reason: result.reason, replySnippet: snippet, activeSince: srv.active_since, inactiveSince: srv.inactive_since, gameStartedAt: srv.game_started_at, gameEndedAt: srv.game_ended_at }; } } } await Promise.all(Array.from({ length: STATUS_CONCURRENCY }, pump)); emit({ type: 'server-status', entries }); } function safeJson(raw: string): T | null { try { return JSON.parse(raw) as T; } catch { return null; } } export function startWorker(): void { if (started) return; started = true; void tick(); void fanoutStatus(); tickTimer = setInterval(tick, TICK_MS); statusTimer = setInterval(fanoutStatus, STATUS_REFRESH_MS); console.log('[worker] started — sweep every', TICK_MS, 'ms, status every', STATUS_REFRESH_MS, 'ms'); } export function stopWorker(): void { if (tickTimer) clearInterval(tickTimer); if (statusTimer) clearInterval(statusTimer); tickTimer = null; statusTimer = null; started = false; } /** * External trigger — called from `/api/push` and beta endpoints so a sweep * fires immediately after a state change instead of waiting for the next tick. */ export async function kick(): Promise { await tick(); }