import { randomUUID } from 'node:crypto'; import type { WebSocket } from 'ws'; import type { AgentAction, AgentStatus, AgentView, LogEntry, LogLevel, ServerToAgent, ServerToDashboard, } from '@stream-control/shared'; import { emptyStatus } from '@stream-control/shared'; import { config } from './config.ts'; import { agentsRepo, logsRepo, type AgentRecord } from './db.ts'; interface PendingCommand { resolve: (value: unknown) => void; reject: (reason: Error) => void; timer: NodeJS.Timeout; } interface AgentConnection { socket: WebSocket; lastSeenAt: number; pending: Map; } /** * Point central : garde en mémoire les connexions agents + le dernier statut * connu, dispatche les commandes et diffuse l'état aux dashboards ouverts. */ class Hub { private readonly connections = new Map(); private readonly statuses = new Map(); private readonly dashboards = new Set(); // --- Agents ------------------------------------------------------------- attachAgent(agentId: string, socket: WebSocket): void { // Une seule session par agent : la nouvelle connexion évince l'ancienne. const existing = this.connections.get(agentId); if (existing && existing.socket !== socket) { this.failPending(existing, new Error('Connexion agent remplacée')); try { existing.socket.close(4000, 'Remplacé par une nouvelle session'); } catch { /* ignore */ } } this.connections.set(agentId, { socket, lastSeenAt: Date.now(), pending: new Map() }); this.statuses.set(agentId, this.statuses.get(agentId) ?? emptyStatus()); this.publishAgent(agentId); } detachAgent(agentId: string, socket: WebSocket): void { const connection = this.connections.get(agentId); if (!connection || connection.socket !== socket) return; this.failPending(connection, new Error('Agent déconnecté')); this.connections.delete(agentId); // On garde le dernier statut connu mais on marque OBS comme injoignable. const status = this.statuses.get(agentId); if (status) { this.statuses.set(agentId, { ...status, obsConnected: false, recording: false, streaming: false, updatedAt: Date.now(), }); } this.publishAgent(agentId); } isOnline(agentId: string): boolean { return this.connections.has(agentId); } markSeen(agentId: string): void { const connection = this.connections.get(agentId); if (connection) connection.lastSeenAt = Date.now(); } updateStatus(agentId: string, status: AgentStatus): void { this.statuses.set(agentId, status); this.markSeen(agentId); this.publishAgent(agentId); } resolveCommand(agentId: string, requestId: string, ok: boolean, data: unknown, error?: string): void { const pending = this.connections.get(agentId)?.pending.get(requestId); if (!pending) return; this.connections.get(agentId)?.pending.delete(requestId); clearTimeout(pending.timer); if (ok) pending.resolve(data); else pending.reject(new Error(error ?? 'La commande a échoué')); } /** Envoie une commande à un agent et attend son accusé de résultat. */ async sendCommand( agentId: string, action: AgentAction, params?: Record, ): Promise { const connection = this.connections.get(agentId); if (!connection) throw new Error('Agent hors-ligne'); const requestId = randomUUID(); const message: ServerToAgent = { type: 'command', requestId, action, params }; return new Promise((resolve, reject) => { const timer = setTimeout(() => { connection.pending.delete(requestId); reject(new Error(`Timeout : l'agent n'a pas répondu à « ${action} »`)); }, config.commandTimeoutMs); connection.pending.set(requestId, { resolve, reject, timer }); try { connection.socket.send(JSON.stringify(message)); } catch (err) { connection.pending.delete(requestId); clearTimeout(timer); reject(err instanceof Error ? err : new Error(String(err))); } }); } /** Pousse la configuration OBS à un agent connecté (sans attendre de réponse). */ pushConfig(record: AgentRecord): void { const connection = this.connections.get(record.id); if (!connection) return; const message: ServerToAgent = { type: 'config', obs: record.obs, autoConnectObs: record.autoConnectObs, watch: record.watch, }; connection.socket.send(JSON.stringify(message)); } disconnectAgent(agentId: string, reason: string): void { const connection = this.connections.get(agentId); if (!connection) return; try { connection.socket.close(4001, reason); } catch { /* ignore */ } } // --- Vues --------------------------------------------------------------- statusOf(agentId: string): AgentStatus { return this.statuses.get(agentId) ?? emptyStatus(); } view(record: AgentRecord): AgentView { const online = this.isOnline(record.id); return { id: record.id, name: record.name, hostname: record.hostname, platform: record.platform, agentVersion: record.agentVersion, online, lastSeenAt: online ? (this.connections.get(record.id)?.lastSeenAt ?? record.lastSeenAt) : record.lastSeenAt, createdAt: record.createdAt, // Le mot de passe OBS n'est jamais renvoyé au navigateur. obs: { ...record.obs, password: record.obs.password ? '********' : '' }, autoConnectObs: record.autoConnectObs, watch: record.watch, notes: record.notes, status: this.statusOf(record.id), }; } views(): AgentView[] { return agentsRepo.list().map((record) => this.view(record)); } // --- Dashboards --------------------------------------------------------- attachDashboard(socket: WebSocket): void { this.dashboards.add(socket); this.sendTo(socket, { type: 'snapshot', agents: this.views(), logs: logsRepo.recent(200), }); } detachDashboard(socket: WebSocket): void { this.dashboards.delete(socket); } publishAgent(agentId: string): void { const record = agentsRepo.get(agentId); if (!record) { this.broadcast({ type: 'agent.removed', agentId }); return; } this.broadcast({ type: 'agent', agent: this.view(record) }); } publishRemoval(agentId: string): void { this.statuses.delete(agentId); this.broadcast({ type: 'agent.removed', agentId }); } /** Journalise un évènement : persistance + diffusion temps réel. */ log(agentId: string | null, level: LogLevel, message: string, ts = Date.now()): LogEntry { const entry = logsRepo.append(agentId, level, message, ts); this.broadcast({ type: 'log', entry }); if (level === 'error' || level === 'warn') { console.warn(`[${level}] ${entry.agentName ?? 'serveur'} — ${message}`); } return entry; } private broadcast(message: ServerToDashboard): void { const payload = JSON.stringify(message); for (const socket of this.dashboards) { if (socket.readyState === socket.OPEN) socket.send(payload); } } private sendTo(socket: WebSocket, message: ServerToDashboard): void { if (socket.readyState === socket.OPEN) socket.send(JSON.stringify(message)); } private failPending(connection: AgentConnection, error: Error): void { for (const pending of connection.pending.values()) { clearTimeout(pending.timer); pending.reject(error); } connection.pending.clear(); } /** Coupe les agents silencieux : le heartbeat n'arrive plus. */ reapStale(): void { const deadline = Date.now() - config.agentTimeoutMs; for (const [agentId, connection] of this.connections) { if (connection.lastSeenAt < deadline) { this.log(agentId, 'warn', 'Agent silencieux, fermeture de la session'); try { connection.socket.terminate(); } catch { /* ignore */ } this.detachAgent(agentId, connection.socket); } } } pingAll(): void { const payload = JSON.stringify({ type: 'ping', ts: Date.now() } satisfies ServerToAgent); for (const connection of this.connections.values()) { if (connection.socket.readyState === connection.socket.OPEN) { connection.socket.send(payload); } } } } export const hub = new Hub();