import type { MonitoringView } from "@batch-cooking/shared"; import type { Server as SocketIoServer } from "socket.io"; import { requireAdminSocket } from "../middlewares/require-admin-socket.js"; import { getMonitoring } from "../modules/admin/admin-monitoring.service.js"; import { logger } from "./logger.service.js"; /** How often the server re-probes every dependency, in ms — see `MonitoringBroadcastService`. */ const PROBE_INTERVAL_MS = 60_000; /** Every authenticated admin socket joins this one room — a single probe loop serves every open tab, instead of each polling independently. */ const MONITORING_ROOM = "monitoring"; /** Event name every `monitoring:update` push carries — the sole message this service ever emits. */ const UPDATE_EVENT = "monitoring:update"; /** * Pushes the admin monitoring status board to every connected admin over * WebSocket, replacing the old `GET /admin/api/monitoring` polling * endpoint entirely (removed — see `admin.routes.ts`'s doc comment): one * server-side probe loop, shared by every viewer, instead of each browser * tab re-probing Postgres/the API/the intent-service/the LLM worker on its * own timer. * * Owns the *whole* `socket.io` server passed to it (`io.use`, top-level * `"connection"` handler) — fine while this is the server's only * real-time feature; a second one added later should likely get its own * namespace rather than share this room's connection handler. */ export class MonitoringBroadcastService { private readonly _io: SocketIoServer; /** * Most recent probe result — replayed to a newly-joined admin * immediately (see the `"connection"` handler below), so a freshly * opened tab isn't left blank for up to a minute waiting for the next * scheduled tick. `null` until the very first probe (kicked off by * {@link start}) completes. */ private _lastSnapshot: MonitoringView | null = null; /** Handle for the running probe loop, or `null` before {@link start}/after {@link stop}. */ private _intervalHandle: NodeJS.Timeout | null = null; public constructor(io: SocketIoServer) { this._io = io; this._io.use(requireAdminSocket); this._io.on("connection", (socket) => { void socket.join(MONITORING_ROOM); if (this._lastSnapshot) { socket.emit(UPDATE_EVENT, this._lastSnapshot); } }); } /** * Starts the probe loop: one immediate probe (so the very first admin * to connect after a server restart doesn't wait a full minute), then * every {@link PROBE_INTERVAL_MS}. Call once, at server boot * (`server.ts`) — meant to run for the lifetime of the process, same * posture as the tech-step worker's own cron (`server.ts` never calls * {@link stop} today; the process exiting takes the interval with it). */ public start(): void { void this._tick(); this._intervalHandle = setInterval(() => void this._tick(), PROBE_INTERVAL_MS); } /** * Stops the probe loop. Nothing in production calls this today — it * exists so a graceful-shutdown hook could use it later, and so tests * can tear an instance down cleanly instead of leaking a live * `setInterval` past their own lifetime (which would otherwise hold the * whole Mocha process open). */ public stop(): void { if (this._intervalHandle) { clearInterval(this._intervalHandle); this._intervalHandle = null; } } /** Probes every dependency, caches the result, and broadcasts it to {@link MONITORING_ROOM}. */ private async _tick(): Promise { try { this._lastSnapshot = await getMonitoring(); this._io.to(MONITORING_ROOM).emit(UPDATE_EVENT, this._lastSnapshot); } catch (err) { // getMonitoring() itself never rejects — every individual probe is // already bounded/caught (see its own doc comment) — so this is // purely defensive, per the repo's await-in-try/catch rule; not // expected to actually fire. logger.error("Monitoring probe tick failed unexpectedly", { error: err instanceof Error ? err.message : String(err), }); } } }