S2: obs-websocket, Wiedergabepuffer und der !clip-Ausloeser

Eigener obs-websocket-v5-Client (das Protokoll ist klein genug, und das
Projekt bleibt ohne Abhaengigkeiten). Anmeldung, Wiederverbindung mit
wachsender Pause, Ereignisse fuer ReplayBufferSaved und Szenenwechsel.

Der Clipper loest zwei Wege gleichzeitig aus: OBS-Wiedergabepuffer fuer die
Rohdatei in Aufnahmequalitaet, Twitch-Clip fuer die teilbare URL. Faellt
einer aus, ist der andere trotzdem etwas wert. Ein direkt in OBS gedrueckter
Hotkey wird ohne unseren Umweg erkannt und als vollwertiger Ausloeser
behandelt.

Die Pfaduebersetzung Windows -> Container hat eine eigene Datei und eigene
Tests - das ist die Stelle, an der die Pipeline am ehesten scheitert. Dazu
Warten auf den abgeschlossenen Schreibvorgang, bevor die Datei uebernommen
wird.

Zum Testvektor der OBS-Anmeldung: die Spezifikation nennt Salt und Challenge,
aber kein erwartetes Ergebnis. Der Erwartungswert wurde unabhaengig mit
Pythons hashlib aus derselben Beschreibung gerechnet; der Test sagt das im
Kommentar, statt einen amtlichen Vektor vorzutaeuschen.

29 Tests gruen. Die Abnahme von S2 braucht ein laufendes OBS und steht aus.

Ausserdem: ADR-0006 haelt fest, dass alles unter einem Dach laeuft - eine
lokale Weboberflaeche plus Kindprozesse fuer Rechenarbeit. Die Prozessliste
in der Architekturdoku ist entsprechend ersetzt. Der Onboarding-Assistent
ist als Teil von S5 aufgenommen.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013VTGjd92LsDDWF3frd9FTB

apps/ingest/src/clips.ts

@@ -0,0 +1,200 @@
+// Clip-Ausloesung. Scheibe S2 aus docs/07-mvp-plan.md.
+//
+// Zwei Wege gleichzeitig, mit unterschiedlichem Zweck:
+//   OBS-Wiedergabepuffer -> Rohdatei in Aufnahmequalitaet, daraus wird das Short
+//   Twitch-Clip ueber Helix -> teilbare URL, im Kanal auffindbar
+// Faellt einer aus, ist der andere trotzdem etwas wert. Siehe
+// docs/projects/clip-shorts.md.
+
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import { logger } from '#core/logger.ts';
+import { ulid } from '#shared/events.ts';
+import type { Repos } from '#db/repos.ts';
+import type { ObsClient } from '#obs/client.ts';
+import { mapReplayPath, waitForStableFile } from '#obs/replay-path.ts';
+import { helix } from '#twitch/helix.ts';
+import { readSecrets } from '#twitch/secrets.ts';
+
+const log = logger('clipper');
+
+export type ClipTrigger = 'command' | 'hotkey' | 'spike' | 'manual';
+
+export type ClipperConfig = {
+  enabled: boolean;
+  alsoCreateTwitchClip: boolean;
+  cooldownGlobalSeconds: number;
+  cooldownUserSeconds: number;
+  followersOnly: boolean;
+  replayDirHost?: string;
+  replayDirContainer: string;
+  clipsDir: string;
+};
+
+export type ClipperDeps = {
+  cfg: ClipperConfig;
+  repo: Repos;
+  obs: ObsClient;
+  say: (text: string) => boolean;
+  currentStreamId: () => string | null;
+  broadcasterId: () => string | null;
+};
+
+export function createClipper(deps: ClipperDeps) {
+  const { cfg, repo, obs } = deps;
+  let lastGlobal = 0;
+  const lastByUser = new Map<string, number>();
+
+  // Wer den Puffer ausgeloest hat, wissen wir erst, wenn OBS die Datei meldet.
+  // Solange steht der Auftrag hier.
+  const awaitingFile: Array<{ clipId: string; at: number }> = [];
+
+  /** Gibt eine Absage zurueck, wenn gerade nicht ausgeloest werden darf. */
+  function checkAllowed(trigger: ClipTrigger, userId: string | null, now: number): string | null {
+    if (!cfg.enabled) return 'Clips sind ausgeschaltet.';
+    if (trigger === 'command' || trigger === 'spike') {
+      const sinceGlobal = (now - lastGlobal) / 1000;
+      if (sinceGlobal < cfg.cooldownGlobalSeconds) {
+        return `Gerade eben schon geclippt - noch ${Math.ceil(cfg.cooldownGlobalSeconds - sinceGlobal)} s.`;
+      }
+    }
+    if (trigger === 'command' && userId) {
+      const last = lastByUser.get(userId) ?? 0;
+      const sinceUser = (now - last) / 1000;
+      if (sinceUser < cfg.cooldownUserSeconds) {
+        return `Du hast gerade erst geclippt - noch ${Math.ceil(cfg.cooldownUserSeconds - sinceUser)} s.`;
+      }
+    }
+    return null;
+  }
+
+  async function trigger(
+    kind: ClipTrigger, opts: { userId?: string; login?: string; score?: number } = {}
+  ): Promise<{ ok: boolean; reason?: string; clipId?: string }> {
+    const now = Date.now();
+    const denied = checkAllowed(kind, opts.userId ?? null, now);
+    if (denied) {
+      log.debug(`Ausloesung abgelehnt (${kind}): ${denied}`);
+      return { ok: false, reason: denied };
+    }
+
+    const clipId = ulid();
+    const streamId = deps.currentStreamId();
+    lastGlobal = now;
+    if (opts.userId) lastByUser.set(opts.userId, now);
+
+    repo.insertClip({
+      id: clipId, streamId, trigger: kind,
+      triggerScore: opts.score ?? null,
+      source: 'obs-replay', status: 'pending',
+    });
+    repo.event('clip.requested', { trigger: kind, by: opts.login ?? null }, streamId,
+      opts.userId ?? null);
+
+    // Beide Wege anstossen, keiner blockiert den anderen
+    const obsOk = saveReplay(clipId);
+    if (cfg.alsoCreateTwitchClip) void createTwitchClip(clipId);
+
+    if (!(await obsOk)) {
+      // Der Twitch-Clip kann trotzdem klappen - der Eintrag bleibt bestehen
+      return { ok: false, reason: 'OBS ist nicht verbunden.', clipId };
+    }
+    return { ok: true, clipId };
+  }
+
+  async function saveReplay(clipId: string): Promise<boolean> {
+    if (!obs.isConnected()) {
+      log.warn('Wiedergabepuffer nicht gesichert: keine OBS-Verbindung');
+      return false;
+    }
+    try {
+      const status = await obs.request<{ outputActive: boolean }>('GetReplayBufferStatus');
+      if (!status.outputActive) {
+        log.warn('Wiedergabepuffer laeuft nicht - in OBS unter Einstellungen -> Ausgabe aktivieren');
+        return false;
+      }
+      awaitingFile.push({ clipId, at: Date.now() });
+      await obs.request('SaveReplayBuffer');
+      return true;
+    } catch (err) {
+      log.error(`SaveReplayBuffer fehlgeschlagen: ${(err as Error).message}`);
+      const i = awaitingFile.findIndex((e) => e.clipId === clipId);
+      if (i !== -1) awaitingFile.splice(i, 1);
+      return false;
+    }
+  }
+
+  async function createTwitchClip(clipId: string): Promise<void> {
+    const broadcasterId = deps.broadcasterId();
+    if (!broadcasterId || !readSecrets().tokens.broadcaster) return;
+    try {
+      const clip = await helix.createClip(broadcasterId);
+      repo.updateClip(clipId, { twitch_clip_id: clip.id });
+      log.info(`Twitch-Clip erstellt: ${clip.id}`);
+    } catch (err) {
+      // Kein harter Fehler: die Rohdatei aus OBS ist die wichtigere Quelle
+      log.warn(`Twitch-Clip nicht erstellt: ${(err as Error).message}`);
+    }
+  }
+
+  /** Wird aufgerufen, wenn OBS eine gespeicherte Pufferdatei meldet. */
+  async function onReplaySaved(hostPath: string): Promise<void> {
+    const containerPath = mapReplayPath(hostPath, cfg.replayDirHost, cfg.replayDirContainer);
+
+    // Auftrag zuordnen. Aeltere als 60 s gehoeren nicht mehr dazu - dann hat
+    // jemand den OBS-Hotkey direkt gedrueckt, ohne unseren Umweg.
+    const now = Date.now();
+    while (awaitingFile.length && now - awaitingFile[0]!.at > 60_000) awaitingFile.shift();
+    const job = awaitingFile.shift();
+
+    let clipId = job?.clipId;
+    if (!clipId) {
+      // Hotkey direkt in OBS gedrueckt - das ist ein vollwertiger Ausloeser
+      clipId = ulid();
+      const streamId = deps.currentStreamId();
+      repo.insertClip({
+        id: clipId, streamId, trigger: 'hotkey', triggerScore: null,
+        source: 'obs-replay', status: 'pending',
+      });
+      repo.event('clip.requested', { trigger: 'hotkey' }, streamId, null);
+      log.info('Wiedergabepuffer per OBS-Hotkey gespeichert');
+      if (cfg.alsoCreateTwitchClip) void createTwitchClip(clipId);
+    }
+
+    try {
+      const size = await waitForStableFile(containerPath);
+      // Vom nur-lesbaren Bind-Mount in den Ablageordner. Ab hier gehoert die
+      // Datei uns, OBS darf seinen Puffer aufraeumen.
+      const dir = path.join(cfg.clipsDir, new Date().toISOString().slice(0, 10), clipId);
+      await fs.mkdir(dir, { recursive: true });
+      const target = path.join(dir, `raw${path.extname(containerPath) || '.mkv'}`);
+      await fs.copyFile(containerPath, target);
+
+      repo.updateClip(clipId, { raw_path: target, status: 'raw' });
+      repo.event('obs.replay_saved', { path: target, bytes: size }, deps.currentStreamId());
+      log.info(`Rohdatei gesichert: ${target} (${(size / 1024 / 1024).toFixed(1)} MB)`);
+      deps.say('Clip gesichert.');
+    } catch (err) {
+      repo.updateClip(clipId, { status: 'failed' });
+      log.error(`Rohdatei nicht uebernommen: ${(err as Error).message}`);
+      log.error(`  OBS meldete: ${hostPath}`);
+      log.error(`  Gesucht unter: ${containerPath}`);
+      log.error('  Stimmt OBS_REPLAY_DIR in der .env mit dem OBS-Zielordner ueberein?');
+    }
+  }
+
+  /** `!clip` im Chat. Gibt die Antwort zurueck, die in den Chat soll. */
+  async function handleCommand(
+    userId: string, login: string, isPrivileged: boolean
+  ): Promise<string | null> {
+    if (cfg.followersOnly && !isPrivileged) {
+      // Follower-Pruefung braucht einen Helix-Aufruf je Nachricht; im MVP
+      // regeln das die Abklingzeiten. Siehe docs/projects/clip-shorts.md.
+    }
+    const res = await trigger('command', { userId, login });
+    if (res.ok) return null;              // Antwort kommt erst, wenn die Datei da ist
+    return res.reason ?? 'Hat nicht geklappt.';
+  }
+
+  return { trigger, onReplaySaved, handleCommand, pendingCount: () => awaitingFile.length };
+}

apps/ingest/src/index.ts

@@ -1,17 +1,20 @@
 // Ingest-Daemon. Liest Chat und Zuschauerzahlen, schreibt beides in die
 // Datenbank und haelt die Einrichtungsseite bereit.
 //
-// Stand S1: IRC + Helix-Polling. EventSub und OBS kommen in S2.
+// Stand S2: IRC, Helix-Polling, obs-websocket und der !clip-Ausloeser.
+// EventSub kommt spaeter - fuer den Clip-Weg reicht IRC plus Polling.
 // Siehe docs/projects/00-ingest.md.
 
-import { loadConfig } from '#core/config.ts';
+import { loadConfig, paths } from '#core/config.ts';
 import { logger } from '#core/logger.ts';
 import { sleep } from '#core/retry.ts';
 import { openDb, closeDb } from '#db/index.ts';
 import { repos } from '#db/repos.ts';
+import { createObs } from '#obs/client.ts';
 import { helix } from '#twitch/helix.ts';
 import { createIrc } from '#twitch/irc.ts';
 import { readSecrets } from '#twitch/secrets.ts';
+import { createClipper } from './clips.ts';
 import { createSpikeDetector } from './spike.ts';
 import { startServer } from './server.ts';
 
@@ -26,12 +29,16 @@ export async function runIngest(): Promise<void> {
   const state = {
     startedAt: new Date().toISOString(),
     streamId: null as string | null,
+    broadcasterId: null as string | null,
     lastSeenLive: 0,
     viewers: 0,
     chatConnected: false,
     chatAnonymous: true,
+    obsConnected: false,
+    obsScene: null as string | null,
     messages: 0,
     spikeCandidates: 0,
+    clipsTriggered: 0,
     lastError: null as string | null,
   };
 
@@ -44,6 +51,39 @@ export async function runIngest(): Promise<void> {
   process.on('SIGINT', () => stop('SIGINT'));
   process.on('SIGTERM', () => stop('SIGTERM'));
 
+  // ---- OBS (Wiedergabepuffer, Szenenwechsel)
+  const obs = createObs(
+    cfg.obs.url,
+    () => readSecrets().obs?.password,
+    {
+      onIdentified: () => { state.obsConnected = true; },
+      onClosed: () => { state.obsConnected = false; },
+      onSceneChanged: (scene) => {
+        state.obsScene = scene;
+        repo.event('obs.scene_changed', { scene }, state.streamId);
+      },
+      onReplaySaved: (hostPath) => { void clipper.onReplaySaved(hostPath); },
+    }
+  );
+
+  const clipper = createClipper({
+    cfg: {
+      enabled: cfg.clipper.enabled,
+      alsoCreateTwitchClip: cfg.clipper.alsoCreateTwitchClip,
+      cooldownGlobalSeconds: cfg.clipper.cooldownGlobalSeconds,
+      cooldownUserSeconds: cfg.clipper.cooldownUserSeconds,
+      followersOnly: cfg.clipper.followersOnly,
+      ...(process.env.OBS_REPLAY_DIR ? { replayDirHost: process.env.OBS_REPLAY_DIR } : {}),
+      replayDirContainer: paths.replay,
+      clipsDir: paths.clips,
+    },
+    repo,
+    obs,
+    say: (text) => irc.say(text),
+    currentStreamId: () => state.streamId,
+    broadcasterId: () => state.broadcasterId,
+  });
+
   startServer({
     channel: cfg.channel.login,
     port: cfg.dashboard.port,
@@ -51,6 +91,7 @@ export async function runIngest(): Promise<void> {
       ...state,
       db: repo.stats(),
       spikeArmed: spikes.isArmed(),
+      clipsWaitingForFile: clipper.pendingCount(),
       note: state.chatAnonymous
         ? 'Chat wird anonym mitgelesen. Zum Senden und fuer Helix: Einrichtung unter /setup.'
         : undefined,
@@ -78,6 +119,12 @@ export async function runIngest(): Promise<void> {
           log.debug(`Erster Stream fuer ${msg.displayName}`);
         }
 
+        if (msg.isCommand && /^!clip\b/i.test(msg.text)) {
+          state.clipsTriggered++;
+          void clipper.handleCommand(msg.userId, msg.displayName, msg.isMod || msg.isBroadcaster)
+            .then((reply) => { if (reply) irc.say(`@${msg.displayName} ${reply}`); });
+        }
+
         const spike = spikes.push(msg);
         if (spike) {
           state.spikeCandidates++;
@@ -93,6 +140,7 @@ export async function runIngest(): Promise<void> {
     },
   });
   void irc.run(abort.signal);
+  void obs.run(abort.signal);
 
   // ---- Zuschauerzahlen und Stream-Sessions
   void pollStream(cfg, repo, state, abort.signal);
@@ -102,7 +150,8 @@ export async function runIngest(): Promise<void> {
 }
 
 type State = {
-  streamId: string | null; lastSeenLive: number; viewers: number; lastError: string | null;
+  streamId: string | null; broadcasterId: string | null;
+  lastSeenLive: number; viewers: number; lastError: string | null;
 };
 
 async function pollStream(
@@ -133,8 +182,10 @@ async function pollStream(
         }
       } else {
         warnedMissingAuth = false;
-        const userId = cfg.channel.userId ||
+        const userId = state.broadcasterId || cfg.channel.userId ||
           (await helix.userByLogin(cfg.channel.login))?.id || '';
+        // Einmal aufgeloest, danach gemerkt - spart einen Aufruf je Minute
+        if (userId) state.broadcasterId = userId;
         const live = userId ? await helix.streamByUserId(userId) : null;
         const now = Date.now();