FP
This commit is contained in:
jeanotx32
267
packages/server/src/hub.ts
Normal file
267
packages/server/src/hub.ts
Normal file
@@ -0,0 +1,267 @@
|
||||
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<string, PendingCommand>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<string, AgentConnection>();
|
||||
private readonly statuses = new Map<string, AgentStatus>();
|
||||
private readonly dashboards = new Set<WebSocket>();
|
||||
|
||||
// --- 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<string, unknown>,
|
||||
): Promise<unknown> {
|
||||
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<unknown>((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,
|
||||
};
|
||||
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,
|
||||
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();
|
||||
Reference in New Issue
Block a user