stream-control/packages/server/src/hub.ts

import { randomUUID } from 'node:crypto';
import type { WebSocket } from 'ws';
import type {
  AgentAction,
  AgentEvent,
  AgentStatus,
  AgentView,
  LogEntry,
  LogLevel,
  ServerToAgent,
  ServerToDashboard,
  WatchTarget,
} from '@stream-control/shared';
import { emptyStatus } from '@stream-control/shared';
import { config } from './config.ts';
import { agentsRepo, logsRepo, spansRepo, targetsRepo, type AgentRecord } from './db.ts';

/**
 * Au-delà de ce silence, une capture en cours est considérée comme terminée.
 *
 * Généreux face aux deux secondes du cycle de statut : un agent qui redémarre
 * pendant un enregistrement doit retrouver son intervalle plutôt que d'en
 * ouvrir un second. Le délai ne coûte rien en justesse — l'intervalle est
 * refermé à sa dernière preuve de vie, pas à l'heure du balayage.
 */
const SPAN_STALE_MS = 60_000;

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.trackRecording(agentId, status);
    this.publishAgent(agentId);
  }

  /**
   * Tient à jour l'intervalle de capture en cours de cette VM.
   *
   * Une pause OBS (show privé) n'interrompt pas l'intervalle : elle n'écrit
   * rien mais ne clôt pas le fichier, et le flux public s'est de toute façon
   * arrêté pendant ce temps — la frise le montre déjà comme une coupure de
   * diffusion. Découper ici laisserait croire à deux captures distinctes.
   */
  private trackRecording(agentId: string, status: AgentStatus): void {
    const now = Date.now();
    const open = spansRepo.openForAgent(agentId);

    if (!status.recording) {
      if (open) spansRepo.close(open.id, now);
      return;
    }

    if (open) {
      // Le profil n'est cherché que tant qu'il manque : une capture déjà
      // rattachée l'est pour de bon, inutile de reposer la question toutes les
      // deux secondes.
      spansRepo.touch(open.id, now, open.targetId ? null : this.recordingTarget(agentId)?.id ?? null);
    } else {
      spansRepo.open(agentId, this.recordingTarget(agentId)?.id ?? null, now);
    }
  }

  /**
   * Le profil que cette VM capture, d'après le pseudo sur lequel sa veille est
   * calée — `startTargetRecording()` l'y pose, c'est le seul lien fiable.
   *
   * Nul si la VM n'enregistre pas, ou si sa veille ne désigne aucun profil
   * suivi : une capture lancée à la main depuis l'onglet Agents, par exemple.
   */
  recordingTarget(agentId: string): WatchTarget | null {
    if (!this.statusOf(agentId).recording) return null;
    const record = agentsRepo.get(agentId);
    if (!record?.watch.username) return null;
    return targetsRepo.findByUsername(record.watch.provider, record.watch.username);
  }

  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>,
    timeoutMs = config.commandTimeoutMs,
  ): 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} »`));
      }, timeoutMs);

      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,
      browser: record.browser,
      recording: record.recording,
    };
    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,
      browser: record.browser,
      recording: record.recording,
      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),
      targets: targetsRepo.list(),
    });
  }

  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 });
  }

  // --- Profils surveillés -------------------------------------------------

  publishTarget(target: WatchTarget): void {
    this.broadcast({ type: 'target', target });
  }

  publishTargetRemoval(targetId: string): void {
    this.broadcast({ type: 'target.removed', targetId });
  }

  /** Passage en direct : le dashboard en fait une notification. */
  publishTargetLive(target: WatchTarget): void {
    this.broadcast({ type: 'target.live', target });
  }

  /** Journalise un évènement : persistance + diffusion temps réel. */
  log(
    agentId: string | null,
    level: LogLevel,
    message: string,
    ts = Date.now(),
    event: AgentEvent | null = null,
  ): LogEntry {
    const entry = logsRepo.append(agentId, level, message, ts, event);
    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 {
    // Même ménage côté captures : une VM disparue en plein enregistrement
    // laisse un intervalle ouvert. On le referme à sa dernière preuve de vie,
    // pas à maintenant — sans quoi une VM éteinte une semaine passerait pour
    // avoir enregistré une semaine.
    spansRepo.sweepStale(Date.now() - SPAN_STALE_MS);

    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();