stream-control/packages/agent/src/index.ts

#!/usr/bin/env node
import os from 'node:os';
import { WebSocket } from 'ws';
import type {
  AgentAction,
  AgentStatus,
  AgentToServer,
  LogLevel,
  ServerToAgent,
  WatchSettings,
} from '@stream-control/shared';
import {
  DEFAULT_WATCH_SETTINGS,
  PROTOCOL_VERSION,
  detectPlatform,
  emptyStatus,
  normalizeWatchSettings,
  safeJsonParse,
} from '@stream-control/shared';
import { loadConfig, persistIdentity, type AgentConfig } from './config.ts';
import { runDiagnostics } from './doctor.ts';
import { ObsController } from './obs.ts';
import { StreamWatcher } from './watcher.ts';
import { cpuUsagePercent, diskUsage, memoryUsage } from './system.ts';

const AGENT_VERSION = '0.1.0';
const RECONNECT_MIN_MS = 1000;
const RECONNECT_MAX_MS = 30_000;

let config: AgentConfig;
try {
  config = loadConfig();
} catch (err) {
  console.error(err instanceof Error ? err.message : err);
  process.exit(1);
}

const obs = new ObsController(config.obs);

const watcher = new StreamWatcher(DEFAULT_WATCH_SETTINGS, {
  recordState: () => obs.recordState(),
  pauseRecording: async () => {
    await obs.execute('record.pause');
  },
  resumeRecording: async () => {
    await obs.execute('record.resume');
  },
});

let socket: WebSocket | null = null;
let statusTimer: NodeJS.Timeout | null = null;
let reconnectDelay = RECONNECT_MIN_MS;
let statusIntervalMs = 2000;
let shuttingDown = false;

// --- Transport vers le serveur de contrôle ----------------------------------

function send(message: AgentToServer): void {
  if (socket?.readyState === WebSocket.OPEN) {
    socket.send(JSON.stringify(message));
  }
}

function report(level: LogLevel, message: string): void {
  const prefix = level === 'error' ? '✖' : level === 'warn' ? '!' : '·';
  console.log(`${prefix} ${message}`);
  send({ type: 'log', level, message, ts: Date.now() });
}

obs.on('log', (level: LogLevel, message: string) => report(level, message));
watcher.on('log', (level: LogLevel, message: string) => report(level, message));

function connect(): void {
  if (shuttingDown) return;

  const url = new URL(config.serverUrl);
  console.log(`Connexion au serveur ${url.origin}${url.pathname}…`);

  socket = new WebSocket(url, {
    headers: { authorization: `Bearer ${config.token}` },
    rejectUnauthorized: !config.insecureTls,
    handshakeTimeout: 10_000,
  });

  socket.on('open', () => {
    reconnectDelay = RECONNECT_MIN_MS;
    send({
      type: 'hello',
      protocol: PROTOCOL_VERSION,
      agentId: config.agentId,
      name: config.name,
      hostname: os.hostname(),
      platform: detectPlatform(process.platform),
      agentVersion: AGENT_VERSION,
    });
  });

  socket.on('message', (raw) => {
    const message = safeJsonParse<ServerToAgent>(raw.toString());
    if (message) void handleServerMessage(message);
  });

  socket.on('close', (code, reason) => {
    stopStatusLoop();
    socket = null;
    if (shuttingDown) return;
    const why = reason.toString() || `code ${code}`;
    console.warn(`Session serveur fermée (${why}), nouvelle tentative dans ${reconnectDelay / 1000}s`);
    scheduleReconnect();
  });

  socket.on('error', (err: Error) => {
    console.error(`Erreur de connexion : ${err.message}`);
  });
}

function scheduleReconnect(): void {
  const delay = reconnectDelay;
  reconnectDelay = Math.min(reconnectDelay * 2, RECONNECT_MAX_MS);
  setTimeout(connect, delay).unref?.();
}

// --- Traitement des messages serveur ----------------------------------------

async function handleServerMessage(message: ServerToAgent): Promise<void> {
  switch (message.type) {
    case 'welcome': {
      statusIntervalMs = message.statusIntervalMs || statusIntervalMs;

      if (message.token && message.token !== config.token) {
        config.token = message.token;
        config.agentId = message.agentId;
        persistIdentity(message.agentId, message.token);
      } else if (!config.agentId) {
        config.agentId = message.agentId;
      }

      obs.applySettings(message.obs);
      applyWatchSettings(message.watch);
      report('info', `Agent « ${config.name} » rattaché au serveur (id ${message.agentId})`);

      if (message.autoConnectObs && !obs.isConnected) {
        obs.connect().catch((err: Error) => report('warn', err.message));
      }
      startStatusLoop();
      break;
    }

    case 'config': {
      obs.applySettings(message.obs);
      applyWatchSettings(message.watch);
      if (message.autoConnectObs && !obs.isConnected) {
        obs.connect().catch((err: Error) => report('warn', err.message));
      }
      break;
    }

    case 'command': {
      try {
        const data = await runAction(message.action, message.params ?? {});
        send({ type: 'result', requestId: message.requestId, ok: true, data });
        void pushStatus(); // état rafraîchi immédiatement après l'action
      } catch (err) {
        const text = err instanceof Error ? err.message : String(err);
        send({ type: 'result', requestId: message.requestId, ok: false, error: text });
        report('error', `Échec de « ${message.action} » : ${text}`);
      }
      break;
    }

    case 'ping': {
      send({ type: 'pong', ts: Date.now() });
      break;
    }
  }
}

/**
 * Aiguille une action : la surveillance et le clavier sont gérés par l'agent,
 * tout le reste part vers obs-websocket.
 */
async function runAction(action: AgentAction, params: Record<string, unknown>): Promise<unknown> {
  switch (action) {
    case 'watch.check':
      return watcher.checkNow();
    case 'hotkey.fullscreen':
      return watcher.restoreFullscreen();
    default:
      return obs.execute(action, params);
  }
}

function applyWatchSettings(raw: WatchSettings | undefined): void {
  watcher.applySettings(normalizeWatchSettings(raw ?? DEFAULT_WATCH_SETTINGS));
}

// --- Boucle de statut --------------------------------------------------------

async function buildStatus(): Promise<AgentStatus> {
  const snapshot = await obs.snapshot();
  const memory = memoryUsage();
  const disk = await diskUsage(snapshot.recordDirectory);

  return {
    ...emptyStatus(),
    ...snapshot,
    watch: watcher.snapshot,
    lastRecordingPath: obs.recordingPath,
    systemCpu: cpuUsagePercent(),
    systemMemoryUsed: memory.used,
    systemMemoryTotal: memory.total,
    diskFreeBytes: disk?.freeBytes,
    diskTotalBytes: disk?.totalBytes,
    updatedAt: Date.now(),
  };
}

async function pushStatus(): Promise<void> {
  if (socket?.readyState !== WebSocket.OPEN) return;
  try {
    send({ type: 'status', status: await buildStatus() });
  } catch (err) {
    console.error('Collecte de statut en échec :', err);
  }
}

function startStatusLoop(): void {
  stopStatusLoop();
  void pushStatus();
  statusTimer = setInterval(() => void pushStatus(), statusIntervalMs);
  statusTimer.unref?.();
}

function stopStatusLoop(): void {
  if (statusTimer) clearInterval(statusTimer);
  statusTimer = null;
}

// --- Cycle de vie ------------------------------------------------------------

async function shutdown(signal: string): Promise<void> {
  if (shuttingDown) return;
  shuttingDown = true;
  console.log(`\n${signal} reçu, arrêt de l'agent…`);
  stopStatusLoop();
  watcher.stop();
  // L'enregistrement OBS en cours n'est volontairement pas interrompu.
  await obs.disconnect().catch(() => undefined);
  socket?.close(1000, 'Arrêt de l\'agent');
  setTimeout(() => process.exit(0), 500).unref();
}

process.on('SIGINT', () => void shutdown('SIGINT'));
process.on('SIGTERM', () => void shutdown('SIGTERM'));
process.on('unhandledRejection', (reason) => {
  console.error('Rejet non géré :', reason);
});

console.log(`stream-control agent v${AGENT_VERSION} — ${config.name} (${process.platform})`);

if (process.argv.includes('--check')) {
  // Diagnostic seul : rien n'est démarré, aucune connexion n'est maintenue.
  runDiagnostics(config).then((code) => process.exit(code));
} else {
  connect();
}