Files
stream-control/packages/agent/src/index.ts
jeanotx32 29645c04ca
All checks were successful
release / build (push) Successful in 21s
release / verify-windows (push) Successful in 1m24s
fix : various stuff
2026-08-12 17:18:36 +02:00

662 lines
24 KiB
JavaScript

#!/usr/bin/env node
import os from 'node:os';
import { WebSocket } from 'ws';
import type {
AgentAction,
AgentEvent,
AgentStatus,
AgentToServer,
FullscreenSettings,
LogLevel,
BrowserSettings,
PresetApplyResult,
RecordingSettings,
ServerToAgent,
WatchSettings,
} from '@stream-control/shared';
import {
DEFAULT_BROWSER_SETTINGS,
DEFAULT_RECORDING_SETTINGS,
DEFAULT_WATCH_SETTINGS,
PROTOCOL_VERSION,
detectPlatform,
emptyStatus,
isRecordingEncoder,
normalizeBrowserSettings,
normalizeRecordingSettings,
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 { currentBuildId, runningBundlePath, selfUpdate } from './updater.ts';
import { blankPage, closeWindow, delay as sleep, openUrl } from './browser.ts';
import { FirefoxController, type StreamQualityOutcome } from './firefox.ts';
import { describeFullscreen, type FullscreenOutcome } from './fullscreen.ts';
import { sendHotkey } from './hotkey.ts';
import { cpuUsagePercent, diskUsage, memoryUsage } from './system.ts';
const AGENT_VERSION = '0.1.0';
/** Empreinte du bundle courant, calculée une fois au démarrage. */
const BUILD_ID = currentBuildId();
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');
},
// Clôture, et non simple `record.stop` : la fenêtre du navigateur doit se
// fermer aussi, sans quoi la VM resterait sur une page morte.
stopRecording: async () => {
await stopCapture();
},
restoreFullscreen: (settings) => restoreFullscreen(settings),
});
let browserSettings: BrowserSettings = DEFAULT_BROWSER_SETTINGS;
/**
* Instance Firefox pilotée, créée à la volée.
*
* Elle n'existe que dans le mode `bidi` : la construire d'office lancerait un
* navigateur sur toutes les VM, y compris celles où l'opérateur ouvre les pages
* lui-même.
*/
let firefox: FirefoxController | null = null;
let recordingSettings: RecordingSettings = DEFAULT_RECORDING_SETTINGS;
/** Un preset attend d'être appliqué : OBS était injoignable ou occupé. */
let presetPending = false;
let socket: WebSocket | null = null;
let statusTimer: NodeJS.Timeout | null = null;
let reconnectTimer: 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));
}
}
/**
* Journalise localement et vers le serveur. `event` classe l'entrée dans
* l'historique de la VM ; l'omettre reste correct pour ce qui n'est que du
* bavardage de progression.
*/
function report(level: LogLevel, message: string, event?: AgentEvent): void {
const prefix = level === 'error' ? '✖' : level === 'warn' ? '!' : '·';
console.log(`${prefix} ${message}`);
send({ type: 'log', level, message, ts: Date.now(), event });
}
obs.on('log', (level: LogLevel, message: string, event?: AgentEvent) =>
report(level, message, event),
);
watcher.on('log', (level: LogLevel, message: string, event?: AgentEvent) =>
report(level, message, event),
);
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);
// Surtout pas de .unref() ici : pendant une coupure, ce minuteur est la seule
// chose qui maintienne la boucle d'évènements en vie. Déréférencé, le process
// sortirait avec le code 0 au lieu de retenter — la reconnexion ne servirait
// jamais. On garde plutôt la référence pour l'annuler à l'arrêt.
reconnectTimer = setTimeout(() => {
reconnectTimer = null;
connect();
}, delay);
}
// --- 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);
applyBrowserSettings(message.browser);
applyRecordingSettings(message.recording);
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);
applyBrowserSettings(message.browser);
applyRecordingSettings(message.recording);
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}`, 'command.failed');
}
break;
}
case 'ping': {
send({ type: 'pong', ts: Date.now() });
break;
}
}
}
function requireUrl(params: Record<string, unknown>): string {
const url = params.url;
if (typeof url !== 'string' || !url.trim()) throw new Error('Paramètre « url » manquant');
return url.trim();
}
/**
* Séquence complète : ouvrir la page, laisser le lecteur démarrer, passer en
* plein écran, régler la qualité, lancer l'enregistrement. Chaque étape est
* journalisée séparément pour qu'un échec désigne son maillon. L'ordre entre
* plein écran et qualité n'est pas arbitraire — voir le commentaire sur
* l'appel à {@link selectMaxStreamQuality}.
*/
async function startCapture(params: Record<string, unknown>): Promise<unknown> {
if (!browserSettings.enabled) {
throw new Error('Pilotage du navigateur désactivé sur cet agent');
}
const url = requireUrl(params);
const opened = await openPage(url);
const wait = Number(params.readyDelayMs ?? browserSettings.readyDelayMs);
report('info', `Attente de ${Math.round(wait / 1000)} s avant le plein écran`);
await sleep(wait);
const fullscreen = watcher.snapshotSettings.fullscreen;
let fullscreenResult: unknown = 'ignoré';
if (fullscreen.enabled) {
// Un échec ici ne doit pas empêcher l'enregistrement : mieux vaut capturer
// une fenêtre non maximisée que ne rien capturer du tout.
fullscreenResult = await restoreFullscreen(fullscreen)
.then((outcome) => {
// `skipped` prime sur `confirmed` : un envoi non renvoyé n'est jamais
// un échec de ce côté-ci, quel que soit le sort de celui qu'il a
// réutilisé.
const failed = !outcome.skipped && outcome.confirmed === false;
report(
failed ? 'warn' : 'info',
describeFullscreen(outcome, fullscreen.key),
failed ? 'fullscreen.failed' : 'fullscreen.restored',
);
return outcome;
})
.catch((err: Error) => {
report('warn', `Plein écran impossible : ${err.message}`, 'fullscreen.failed');
return `échec : ${err.message}`;
});
}
// Après, et non avant, le plein écran : celui-ci envoie une touche au
// lecteur, et la plupart des lecteurs vidéo traitent n'importe quelle touche
// comme une activité qui réaffiche les contrôles — y compris une touche sans
// rapport avec la qualité. Dans l'ordre inverse, le curseur écarté en fin de
// sélection de qualité (pour laisser la surbrillance des contrôles
// s'estomper) se faisait donc aussitôt annuler par cette touche. Ici,
// l'éloignement du curseur reste la dernière chose à toucher la page avant
// l'enregistrement.
await selectMaxStreamQuality();
// Dernière chance d'écrire un preset en attente : une fois l'enregistrement
// commencé, OBS refuse toute modification jusqu'à l'arrêt. Sans cet appel
// explicite, l'écriture dépendait du hasard d'un cycle de statut (toutes les
// 2 s) tombant avant le démarrage — perdu, le preset restait en attente
// jusqu'au prochain arrêt, et cet enregistrement tournait sur l'ancien réglage.
await drainPresetApply();
await obs.execute('record.start');
report('info', `Capture démarrée pour ${url}`, 'capture.started');
return { opened, fullscreen: fullscreenResult, recording: true };
}
/** Ouvre une page par la voie correspondant au mode de pilotage. */
async function openPage(url: string): Promise<unknown> {
const bidi = driver();
if (bidi) return bidi.navigate(url);
return openUrl(browserSettings, url, report, {
forceXWayland: watcher.snapshotSettings.fullscreen.enabled,
});
}
/** Décharge ou ferme la fenêtre, selon le réglage et le mode de pilotage. */
async function releasePage(): Promise<unknown> {
const bidi = driver();
if (bidi) return browserSettings.onStop === 'close' ? bidi.quit() : bidi.blank();
return browserSettings.onStop === 'close'
? closeWindow(watcher.snapshotSettings.fullscreen.windowMatch, report)
: blankPage(browserSettings, report, {
forceXWayland: watcher.snapshotSettings.fullscreen.enabled,
});
}
async function stopCapture(): Promise<unknown> {
const result = await obs.execute('record.stop');
// Fermer la fenêtre détruit la source « capture de fenêtre » d'OBS, qu'il faut
// ensuite repointer à la main. D'où le défaut « page vide », qui décharge le
// lecteur sans faire disparaître la fenêtre.
let closed: unknown = 'conservée';
if (browserSettings.enabled && browserSettings.onStop !== 'keep') {
closed = await releasePage().catch((err: Error) => {
report('warn', `Libération de la fenêtre impossible : ${err.message}`, 'command.failed');
return `échec : ${err.message}`;
});
}
report('info', 'Capture arrêtée', 'capture.stopped');
return { ...(result as object), window: closed };
}
/**
* Aiguille une action : navigateur, surveillance et clavier sont gérés par
* l'agent lui-même ; 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();
case 'browser.open':
return openPage(requireUrl(params));
case 'browser.close':
return driver()?.quit() ?? closeWindow(watcher.snapshotSettings.fullscreen.windowMatch, report);
case 'browser.importSession': {
const bidi = driver();
if (!bidi) {
throw new Error(
"L'import de session exige le mode de pilotage « WebDriver BiDi » sur cet agent",
);
}
return bidi.importSession();
}
case 'capture.start':
return startCapture(params);
case 'capture.stop':
return stopCapture();
case 'record.start':
// Même raison que dans startCapture() : sans ce passage explicite, un
// preset changé juste avant restait en attente jusqu'au prochain arrêt,
// et cette capture démarrait sur l'ancien réglage.
await drainPresetApply();
return obs.execute('record.start');
case 'preset.apply':
return applyPresetNow(params);
case 'agent.update':
return selfUpdate(
typeof params.url === 'string' && params.url ? params.url : config.packageUrl,
{
recordState: () => obs.recordState(),
log: (level, message, event) => report(level, message, event),
},
);
default:
return obs.execute(action, params);
}
}
function applyWatchSettings(raw: WatchSettings | undefined): void {
watcher.applySettings(normalizeWatchSettings(raw ?? DEFAULT_WATCH_SETTINGS));
}
function applyBrowserSettings(raw: BrowserSettings | undefined): void {
browserSettings = normalizeBrowserSettings(raw ?? DEFAULT_BROWSER_SETTINGS);
if (browserSettings.mode !== 'bidi') {
// On repasse en mode `launch` : le canal n'a plus de sens, mais on laisse
// volontairement la fenêtre ouverte — OBS la capture peut-être encore.
firefox?.dispose();
firefox = null;
return;
}
if (firefox) firefox.applySettings(browserSettings);
else firefox = new FirefoxController(browserSettings, report);
}
/** Contrôleur BiDi, ou `null` si l'agent n'est pas dans ce mode. */
function driver(): FirefoxController | null {
return browserSettings.mode === 'bidi' ? firefox : null;
}
/**
* Au-delà de ce délai depuis le dernier envoi, un nouveau rappel est à
* nouveau autorisé. Choisi généreux face au 1200 ms que `restoreFullscreen`
* attend en interne avant de relire l'état : le but n'est pas de couvrir
* l'animation, mais la fenêtre où le flux public vient tout juste de démarrer
* et peut encore vaciller (bref retour en privé, requalification du statut).
*/
const FULLSCREEN_COOLDOWN_MS = 10_000;
/** Aucun rappel encore envoyé sur cette instance d'agent. */
let lastFullscreenAt = 0;
/** Deux appels concurrents doivent partager le même envoi, pas en déclencher deux. */
let fullscreenInFlight: Promise<FullscreenOutcome> | null = null;
/**
* Rappelle le plein écran par la voie correspondant au mode de pilotage.
*
* En BiDi, la demande part dans Firefox et son effet est relu dans la page. En
* `launch`, on en reste à une touche envoyée au serveur d'affichage, sans
* moyen de savoir ce qu'elle a produit — d'où le `confirmed` absent.
*
* Point de passage unique, et c'est voulu : deux mécanismes indépendants
* peuvent réclamer ce rappel — la séquence d'ouverture ({@link startCapture})
* et {@link StreamWatcher}, qui le redemande de lui-même après un show privé.
* Sans coordination, un flux qui vacille juste après le démarrage (un retour
* bref en privé pendant que la séquence d'ouverture patiente encore) fait
* partir les deux envois à quelques secondes d'écart. La touche du lecteur
* basculant l'affichage plutôt que le forçant, le second envoi annule le
* premier au lieu de le confirmer — d'où une fenêtre qui, au lieu de
* s'agrandir, finit par rétrécir. Le garde-fou vit ici et non chez l'un des
* deux appelants : c'est le seul endroit que les deux traversent forcément.
*/
async function restoreFullscreen(settings: FullscreenSettings): Promise<FullscreenOutcome> {
// Un envoi est déjà en cours : partager son résultat, pas en déclencher un
// second. Marqué `skipped` malgré le succès qu'il rapporte — c'est cette
// marque, et non la réussite, qui doit gouverner le message affiché à
// l'appelant (voir describeFullscreen) : sans elle, les deux appelants
// journalisent chacun « touche envoyée » pour un seul envoi réel, ce qui
// se lit comme deux rappels distincts alors qu'il n'y en a eu qu'un.
if (fullscreenInFlight) {
const outcome = await fullscreenInFlight;
return { ...outcome, skipped: true };
}
const sinceLast = Date.now() - lastFullscreenAt;
if (lastFullscreenAt > 0 && sinceLast < FULLSCREEN_COOLDOWN_MS) {
return { method: 'ignoré (rappel trop rapproché)', skipped: true };
}
const bidi = driver();
fullscreenInFlight = (
bidi ? bidi.restoreFullscreen(settings) : sendHotkey({ key: settings.key, windowMatch: settings.windowMatch })
).finally(() => {
fullscreenInFlight = null;
lastFullscreenAt = Date.now();
});
return fullscreenInFlight;
}
/**
* Sélectionne la meilleure qualité offerte par le lecteur Stripchat.
*
* Sans effet hors mode BiDi : il n'existe aucun moyen d'interagir avec le
* contenu de la page sans lui. Un échec ne bloque jamais l'enregistrement —
* mieux vaut capturer une qualité non maximale que ne rien capturer du tout.
*/
async function selectMaxStreamQuality(): Promise<void> {
const bidi = driver();
const settings = watcher.snapshotSettings.streamQuality;
if (!bidi || !settings.enabled) return;
await sleep(settings.delayMs);
const result: StreamQualityOutcome = await bidi
.selectMaxStreamQuality()
.catch((err: Error) => ({ ok: false, reason: err.message }));
if (!result.ok) {
// Le diagnostic (menu ouvert mais balisage inattendu) part dans le même
// message : plus vite exploitable que de redemander une capture DevTools.
const suffix = result.debug ? `${result.debug}` : '';
report('warn', `Qualité du flux non réglée : ${result.reason}${suffix}`, 'quality.failed');
return;
}
report(
'info',
result.already
? `Qualité du flux déjà au maximum (${result.label})`
: `Qualité du flux réglée sur ${result.label}`,
'quality.selected',
);
}
// --- Presets d'enregistrement ------------------------------------------------
/**
* Enregistre le preset voulu. L'application effective est différée : OBS peut
* être déconnecté ou en pleine capture au moment où la configuration arrive.
*/
function applyRecordingSettings(raw: RecordingSettings | undefined): void {
const next = normalizeRecordingSettings(raw ?? DEFAULT_RECORDING_SETTINGS);
const changed =
next.enabled !== recordingSettings.enabled ||
next.presetId !== recordingSettings.presetId ||
next.encoder !== recordingSettings.encoder;
recordingSettings = next;
if (next.enabled && changed) presetPending = true;
}
/**
* Applique le preset en attente dès que les conditions le permettent.
*
* Un échec ne relance pas la tentative : une clé refusée par cette version d'OBS
* le sera à chaque cycle, et marteler l'API n'y changerait rien. Le prochain
* enregistrement de la configuration, ou le bouton du dashboard, réessaiera.
*/
async function drainPresetApply(): Promise<void> {
if (!presetPending || !recordingSettings.enabled || !obs.isConnected) return;
const state = await obs.recordState().catch(() => null);
if (!state || state.active) return; // jamais pendant une capture
presetPending = false;
try {
report(
'info',
describePreset(await obs.applyRecordingPreset(recordingSettings)),
'preset.applied',
);
} catch (err) {
report(
'warn',
`Preset non appliqué : ${err instanceof Error ? err.message : String(err)}`,
'preset.failed',
);
}
}
/** Application à la demande : passe outre l'interrupteur, l'opérateur a tranché. */
async function applyPresetNow(params: Record<string, unknown>): Promise<PresetApplyResult> {
const result = await obs.applyRecordingPreset({
enabled: true,
presetId:
typeof params.presetId === 'string' && params.presetId
? params.presetId
: recordingSettings.presetId,
encoder: isRecordingEncoder(params.encoder) ? params.encoder : recordingSettings.encoder,
});
presetPending = false;
report('info', describePreset(result), 'preset.applied');
return result;
}
function describePreset(result: PresetApplyResult): string {
const parts = [
`Preset « ${result.presetLabel} » appliqué — ${result.encoder}, ${result.video}`,
];
if (result.skipped.length > 0) parts.push(`refusé par OBS : ${result.skipped.join(', ')}`);
if (result.restartRequired) {
parts.push("l'encodeur a changé : redémarre OBS pour qu'il le charge");
}
return parts.join(' — ');
}
// --- 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,
browser: driver()?.state,
buildId: BUILD_ID,
canSelfUpdate: Boolean(runningBundlePath() && config.packageUrl),
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 {
// Le cycle de statut sert aussi d'horloge au preset en attente : il tourne
// déjà, et il s'exécute justement quand OBS vient de redevenir joignable.
await drainPresetApply();
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();
if (reconnectTimer) clearTimeout(reconnectTimer);
reconnectTimer = null;
watcher.stop();
// Le canal BiDi se ferme, mais pas Firefox : sa fenêtre est la source de
// capture d'OBS, et l'enregistrement en cours doit lui survivre.
firefox?.dispose();
// 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();
}