abdulive/apps/ingest/src/fern.ts

// Fernsteuerung fuer OBS, Seite des Stream-PCs.
//
// Das Pult am Handy redet mit dem Hub, OBS laeuft hier. Der Hub kommt an
// diesen Rechner nicht heran (fremder Router, keine Portfreigabe), also waehlt
// sich der PC beim Hub ein und haelt die Leitung offen:
//
//   GET  /fern/befehle   SSE, Befehle kommen herein     (Hub -> PC)
//   POST /fern/stand     Szenen, aktuelle Szene, Ton    (PC -> Hub)
//   POST /fern/antwort   ob ein Befehl geklappt hat     (PC -> Hub)
//
// Gegenstueck: apps/overlay-server/src/fern.ts. Eingerichtet wird mit
// `abdulive fern:verbinden <adresse> <schluessel>`; beides steht fertig auf
// der Uebersichtsseite des Hubs.

import { logger } from '#core/logger.ts';
import { forever, sleep } from '#core/retry.ts';
import type { ObsClient } from '#obs/client.ts';

const log = logger('fern');

/** Der Hub schickt alle 15 s einen Herzschlag. Drei verpasste heisst: tot. */
const TOT_NACH_MS = 45_000;

/** Nach diesen OBS-Ereignissen stimmt der gemeldete Stand nicht mehr. */
const STAND_EREIGNISSE = new Set([
  'CurrentProgramSceneChanged', 'SceneListChanged', 'SceneCreated', 'SceneRemoved',
  'SceneNameChanged', 'InputMuteStateChanged', 'InputCreated', 'InputRemoved',
  'InputNameChanged', 'ReplayBufferStateChanged',
]);

export type Stand = {
  obs: boolean;
  szenen: string[];
  aktuell: string | null;
  ton: Array<{ name: string; stumm: boolean }>;
  puffer: boolean;
};

export type Befehl =
  | { id: string; art: 'szene'; szene: string }
  | { id: string; art: 'clip' }
  | { id: string; art: 'stumm'; quelle: string; stumm: boolean };

/**
 * Zerlegt einen SSE-Strom in Nachrichten. Bekommt beliebige Stuecke - eine
 * Nachricht kann mitten im Wort geteilt ankommen, und zwei koennen in einem
 * Stueck stecken.
 */
export function sseZerleger(): (stueck: string) => string[] {
  let rest = '';
  return (stueck) => {
    rest += stueck.replace(/\r\n?/g, '\n');
    const fertig: string[] = [];
    let i: number;
    while ((i = rest.indexOf('\n\n')) !== -1) {
      const block = rest.slice(0, i);
      rest = rest.slice(i + 2);
      const daten = block.split('\n')
        .filter((z) => z.startsWith('data:'))
        .map((z) => z.slice(5).replace(/^ /, ''));
      if (daten.length) fertig.push(daten.join('\n'));
    }
    return fertig;
  };
}

/** Was vom Hub kommt, pruefen. Unbekanntes wird nicht ausgefuehrt. */
export function befehlAus(roh: unknown): Befehl | null {
  if (!roh || typeof roh !== 'object') return null;
  const r = roh as Record<string, unknown>;
  const id = typeof r['id'] === 'string' ? r['id'] : '';
  if (!id) return null;
  if (r['art'] === 'szene' && typeof r['szene'] === 'string') return { id, art: 'szene', szene: r['szene'] };
  if (r['art'] === 'clip') return { id, art: 'clip' };
  if (r['art'] === 'stumm' && typeof r['quelle'] === 'string') {
    return { id, art: 'stumm', quelle: r['quelle'], stumm: r['stumm'] === true };
  }
  return null;
}

/** Szenen, aktuelle Szene, stummschaltbare Eingaenge und der Puffer. */
export async function standLesen(obs: ObsClient): Promise<Stand> {
  const leer: Stand = { obs: false, szenen: [], aktuell: null, ton: [], puffer: false };
  if (!obs.isConnected()) return leer;
  try {
    const liste = await obs.request<{
      currentProgramSceneName: string;
      scenes: Array<{ sceneName: string; sceneIndex: number }>;
    }>('GetSceneList');
    // OBS zaehlt von unten. Im Fenster steht die mit dem hoechsten Index oben,
    // und in dieser Reihenfolge erwartet man sie auch am Handy.
    const szenen = [...liste.scenes]
      .sort((a, b) => b.sceneIndex - a.sceneIndex).map((s) => s.sceneName);

    const eingaenge = await obs.request<{ inputs: Array<{ inputName: string }> }>('GetInputList');
    // Welcher Eingang Ton hat, sagt OBS nicht direkt. Wer auf GetInputMute
    // antwortet, hat welchen; Bildquellen lehnen ab.
    const stumm = await Promise.allSettled(eingaenge.inputs.map((e) =>
      obs.request<{ inputMuted: boolean }>('GetInputMute', { inputName: e.inputName })
        .then((r) => ({ name: e.inputName, stumm: r.inputMuted }))));
    const ton = stumm.flatMap((r) => (r.status === 'fulfilled' ? [r.value] : []));

    const puffer = await obs.request<{ outputActive: boolean }>('GetReplayBufferStatus')
      .then((r) => r.outputActive, () => false);

    return { obs: true, szenen, aktuell: liste.currentProgramSceneName || null, ton, puffer };
  } catch (err) {
    log.debug(`Stand nicht gelesen: ${(err as Error).message}`);
    return { ...leer, obs: obs.isConnected() };
  }
}

export type FernDeps = {
  url: string;
  key: string;
  obs: ObsClient;
  /** Den Wiedergabepuffer sichern, wie bei !clip, nur ohne Abklingzeit. */
  clip: () => Promise<{ ok: boolean; reason?: string }>;
};

export type FernAnschluss = {
  run: (signal: AbortSignal) => Promise<void>;
  /** Nach einem OBS-Ereignis rufen. Sammelt kurz, dann geht ein Stand raus. */
  obsEreignis: (type: string) => void;
  /** OBS verbunden oder getrennt. */
  obsWechsel: () => void;
  verbunden: () => boolean;
};

export function createFernanschluss(deps: FernDeps): FernAnschluss {
  const basis = deps.url.replace(/\/+$/, '');
  const kopf = { authorization: `Bearer ${deps.key}` };
  let verbunden = false;
  let sammler: NodeJS.Timeout | null = null;

  async function post(pfad: string, body: unknown): Promise<void> {
    try {
      const res = await fetch(`${basis}${pfad}`, {
        method: 'POST',
        headers: { ...kopf, 'content-type': 'application/json' },
        body: JSON.stringify(body),
        signal: AbortSignal.timeout(10_000),
      });
      if (!res.ok) log.warn(`${pfad}: Hub antwortet ${res.status}`);
    } catch (err) {
      log.debug(`${pfad} nicht zugestellt: ${(err as Error).message}`);
    }
  }

  const standSenden = async (): Promise<void> => {
    if (!verbunden) return;
    await post('/fern/stand', await standLesen(deps.obs));
  };

  const spaeterSenden = (): void => {
    // OBS feuert beim Szenenwechsel mehrere Ereignisse kurz hintereinander.
    // Ein Stand danach reicht.
    if (sammler) clearTimeout(sammler);
    sammler = setTimeout(() => { sammler = null; void standSenden(); }, 300);
  };

  async function ausfuehren(b: Befehl): Promise<{ ok: boolean; text: string }> {
    if (b.art === 'clip') {
      const r = await deps.clip();
      return r.ok ? { ok: true, text: 'Clip wird gesichert.' }
        : { ok: false, text: r.reason ?? 'Clip ging nicht.' };
    }
    if (!deps.obs.isConnected()) return { ok: false, text: 'OBS ist auf dem Stream-PC nicht verbunden.' };
    try {
      if (b.art === 'szene') {
        await deps.obs.request('SetCurrentProgramScene', { sceneName: b.szene });
        return { ok: true, text: `Szene: ${b.szene}` };
      }
      await deps.obs.request('SetInputMute', { inputName: b.quelle, inputMuted: b.stumm });
      return { ok: true, text: `${b.quelle} ${b.stumm ? 'stumm' : 'wieder an'}` };
    } catch (err) {
      return { ok: false, text: (err as Error).message };
    }
  }

  async function einmal(signal: AbortSignal): Promise<void> {
    const wachhund = new AbortController();
    const beides = AbortSignal.any([signal, wachhund.signal]);
    let zuletzt = Date.now();
    const takt = setInterval(() => {
      if (Date.now() - zuletzt > TOT_NACH_MS) {
        log.warn('Vom Hub nichts mehr gehoert, verbinde neu');
        wachhund.abort();
      }
    }, 10_000);

    try {
      const res = await fetch(`${basis}/fern/befehle`, {
        headers: { ...kopf, accept: 'text/event-stream' }, signal: beides,
      });
      if (res.status === 403) {
        log.error('Der Hub lehnt den Schluessel ab. Neu einrichten mit dem Befehl ' +
          'von der Uebersichtsseite: abdulive fern:verbinden ...');
        // Nicht im Sekundentakt anklopfen, der Schluessel wird nicht von allein richtig.
        await sleep(5 * 60_000, signal).catch(() => {});
        throw new Error('Schluessel abgelehnt');
      }
      if (!res.ok || !res.body) throw new Error(`Hub antwortet ${res.status}`);

      verbunden = true;
      log.info(`Mit dem Hub verbunden (${basis}) - das Pult kann OBS jetzt steuern`);
      void standSenden();

      const zerleger = sseZerleger();
      const decoder = new TextDecoder();
      for await (const stueck of res.body as unknown as AsyncIterable<Uint8Array>) {
        zuletzt = Date.now();
        for (const daten of zerleger(decoder.decode(stueck, { stream: true }))) {
          let roh: unknown;
          try { roh = JSON.parse(daten); } catch { continue; }
          const b = befehlAus(roh);
          if (!b) continue;
          void ausfuehren(b).then((a) => {
            log.info(`Pult: ${b.art} -> ${a.ok ? 'ok' : 'abgelehnt'} (${a.text})`);
            return post('/fern/antwort', { id: b.id, ...a });
          }).then(spaeterSenden);
        }
      }
      throw new Error('Hub hat die Leitung geschlossen');
    } catch (err) {
      if (!signal.aborted) log.debug(`Leitung zum Hub weg: ${(err as Error).message}`);
      throw err;
    } finally {
      clearInterval(takt);
      if (verbunden) log.info('Leitung zum Hub getrennt');
      verbunden = false;
    }
  }

  return {
    run: (signal) => forever(einmal, {
      signal, baseMs: 2000, maxMs: 60_000,
      onWait: (ms, n) => {
        const zeile = `Hub nicht erreichbar, neuer Versuch in ${Math.round(ms / 1000)} s`;
        if (n >= 3) log.warn(zeile); else log.debug(zeile);
      },
    }),
    obsEreignis: (type) => { if (STAND_EREIGNISSE.has(type)) spaeterSenden(); },
    obsWechsel: spaeterSenden,
    verbunden: () => verbunden,
  };
}