Hub: der Dienst laeuft zentral statt auf dem Stream-PC

Zwei Dinge, die Reichweite bringen sollen, hingen bisher daran, dass jemand
vor seinem Rechner sitzt: der Discord-Rollen-Knopf und die Go-Live-
Ankuendigung. Beide brauchen nur Internet.

Der Schnitt verlaeuft entlang einer Frage: braucht die Aufgabe eine Datei
oder ein Programm, das nur auf dem Stream-PC existiert?

  Hub (LXC)                    Ingest (Abdus PC)
  ---------                    -----------------
  Chat, Zuschauerzahlen        OBS-Wiedergabepuffer
  Discord: Knopf und Go-Live   !clip, Rendern, Untertitel
  Overlay-Seiten fuer OBS

Beide schreiben in ihre eigene Datenbank. Eine gemeinsame waere eine
Netzwerkfreigabe fuer SQLite - der zuverlaessigste bekannte Weg zu einer
beschaedigten Datei.

Das kostet ADR-0005 seine Hauptbegruendung: "bindet nur auf 127.0.0.1" gilt
fuer den Hub nicht mehr. An die Stelle treten ein Passwort mit signiertem
Sitzungs-Cookie fuer alles, was ein Mensch bedient, und ein zufaelliger
Schluessel in der Adresse fuer die Browser-Sources. Der Schluessel ist kein
schoenes Verfahren - er steht in jedem OBS-Profil -, aber eine Browser-Source
kann kein Formular ausfuellen und keinen Login durchhalten.

`publicUrl` haengt der OAuth-Rueckruf daran. Twitch vergleicht die
Redirect-URI zeichengenau, und ein fehlender Schraegstrich sieht im Fehlertext
aus wie ein falsches Client-Secret.

Die Routen der Einrichtungsseite sind aus ingest/server.ts herausgeloest,
weil sie jetzt an zwei Servern haengen.

Begruendung und Konsequenzen in ADR-0008.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

apps/cli/main.ts

@@ -646,6 +646,177 @@ Was jetzt noch von Hand gehoert:
      verbinden, dann bekommen Subs automatisch eine Rolle.`);
 }
 
+// ---------------------------------------------------------------- overlay
+
+/**
+ * Overlay-Server allein starten. Zum Ansehen und Einrichten, ohne dass der
+ * ganze Hub laeuft. Mit --demo laeuft erfundener Betrieb: anders koennte man
+ * eine Raid-Einblendung nur pruefen, indem man auf einen echten Raid wartet.
+ */
+async function overlayOnly(args: string[]): Promise<void> {
+  const cfg = loadConfig();
+  const { startOverlayServer } = await import('../overlay-server/src/server.ts');
+  const { createBus } = await import('../overlay-server/src/bus.ts');
+
+  const bus = createBus();
+  const abort = new AbortController();
+  process.on('SIGINT', () => { abort.abort(); process.exit(0); });
+
+  startOverlayServer({
+    port: cfg.overlay.port,
+    channel: cfg.channel.login,
+    publicUrl: cfg.publicUrl,
+    bus,
+  });
+
+  if (args.includes('--demo')) {
+    const { startDemo } = await import('../overlay-server/src/demo.ts');
+    startDemo(bus, abort.signal);
+  }
+
+  const wo = cfg.publicUrl || `http://127.0.0.1:${cfg.overlay.port}`;
+  console.log(`
+Uebersicht mit allen Adressen fuer OBS: ${wo}/`);
+  console.log(`${C.dim}Beenden mit Strg+C.${C.off}`);
+  await new Promise(() => { /* laeuft bis Strg+C */ });
+}
+
+/** Passwort fuer die Weboberflaeche setzen. Nur hier, nie ueber das Netz. */
+async function webPasswort(args: string[]): Promise<void> {
+  const { setzePasswort } = await import('../overlay-server/src/auth.ts');
+  const pw = args[0] || process.env.ABDULIVE_WEB_PASSWORT ||
+    await promptHidden('Neues Passwort (Eingabe wird nicht angezeigt): ');
+  setzePasswort(pw.trim());
+  console.log(`${C.ok}Passwort gesetzt.${C.off} Angemeldet bleibt man 30 Tage.`);
+}
+
+// ---------------------------------------------------------------- golive
+
+/**
+ * Probelauf der Go-Live-Ankuendigung.
+ *
+ * Ohne Argument geht der Post in den internen Kanal und pingt niemanden -
+ * damit laesst sich das Aussehen pruefen, ohne den Server aufzuwecken. Mit
+ * `--echt` geht er dorthin, wo er im Betrieb hinginge, mit Ping. Das ist der
+ * einzige Weg, vor dem ersten Stream zu sehen, ob der Ping wirklich ankommt;
+ * eine Erwaehnung ohne `allowed_mentions` sieht naemlich genauso aus wie eine
+ * mit, benachrichtigt aber niemanden.
+ */
+async function goLiveTest(args: string[]): Promise<void> {
+  const echt = args.includes('--echt');
+  const cfg = loadConfig();
+  const { createAnnouncer } = await import('../broadcaster/src/golive.ts');
+  const { repos } = await import('#db/repos.ts');
+  const { helix } = await import('#twitch/helix.ts');
+  const { readSecrets } = await import('#twitch/secrets.ts');
+
+  const discord = readSecrets().discord;
+  const ziel = echt ? discord?.goLiveWebhook : discord?.technikWebhook;
+  if (!ziel) {
+    throw new Error(echt
+      ? 'Kein Go-Live-Webhook hinterlegt. Einmal `abdulive discord:setup` laufen lassen.'
+      : 'Kein Webhook fuer den internen Kanal. Einmal `abdulive discord:setup` laufen lassen ' +
+        '- der Technik-Webhook ist seit Fassung 0.6 im Bauplan.');
+  }
+
+  // Wenn Abdu gerade live ist, wird der echte Stream angekuendigt. Sonst ein
+  // erfundener - das prueft denselben Weg, nur mit anderem Text.
+  let stream = {
+    id: `probe-${Date.now()}`,
+    title: 'Probelauf der Go-Live-Ankuendigung',
+    category: 'Just Chatting',
+    login: cfg.channel.login,
+    startedAt: new Date().toISOString(),
+  } as import('../broadcaster/src/golive.ts').StreamInfo;
+
+  try {
+    const user = await helix.userByLogin(cfg.channel.login);
+    const live = user ? await helix.streamByUserId(user.id) : null;
+    if (live) {
+      stream = {
+        id: `probe-${live.id}-${Date.now()}`,
+        title: live.title,
+        category: live.game_name,
+        login: cfg.channel.login,
+        startedAt: live.started_at,
+        thumbnailUrl: live.thumbnail_url,
+      };
+      console.log(`${C.dim}Stream laeuft gerade - der Probelauf nimmt die echten Daten.${C.off}`);
+    } else {
+      console.log(`${C.dim}Kein laufender Stream - der Probelauf nimmt erfundene Daten.${C.off}`);
+    }
+  } catch (err) {
+    // Haeufigster Fall ist nicht "Netz weg", sondern "noch keine Twitch-App
+    // hinterlegt". Der Satz soll das sagen, sonst sucht man am falschen Ende.
+    console.log(`${C.warn}Twitch nicht abgefragt:${C.off} ${(err as Error).message}`);
+    console.log(`${C.dim}Der Probelauf nimmt erfundene Daten - der Weg zum Webhook ` +
+      `wird trotzdem vollstaendig geprueft.${C.off}`);
+  }
+
+  // OBS nur dazunehmen, wenn es laeuft: sonst wartet der Screenshot ins Leere.
+  let obs: Awaited<ReturnType<typeof withObs>> | null = null;
+  if (cfg.broadcaster.obsScreenshot && await tcpProbe(OBS_HOST, OBS_PORT, 1500)) {
+    obs = await withObs();
+  }
+
+  const announcer = createAnnouncer({
+    cfg: {
+      enabled: true,
+      mode: echt ? 'live' : 'draft',
+      delaySeconds: 0,
+      cooldownHours: cfg.broadcaster.cooldownHours,
+      pingRole: cfg.broadcaster.pingRole,
+      obsScreenshot: Boolean(obs),
+    },
+    repo: repos(openDb()),
+    ...(obs ? { obs: obs.client } : {}),
+    // Ein Probelauf geht durch die Abklingzeit hindurch, sonst koennte man ihn
+    // nach einem echten Stream sechs Stunden lang nicht wiederholen.
+    refresh: async (s) => s,
+  });
+
+  // Als eigene Art protokollieren: sonst sperrt der Probelauf ueber die
+  // Abklingzeit die naechste echte Ankuendigung.
+  const result = await announcer.announceNow(stream, { force: true, art: 'golive-test' });
+  obs?.stop();
+
+  if (result.status === 'gesendet') {
+    const wohin = result.mode === 'draft' ? 'interner Kanal' : 'Ankuendigungen';
+    const ping = result.pinged ? `mit Ping an "${cfg.broadcaster.pingRole}"` : 'ohne Ping';
+    console.log(`
+${C.ok}Post ist raus${C.off} (${wohin}, ${ping}).`);
+    if (result.url) console.log(`  ${result.url}`);
+    if (echt && !result.pinged) {
+      console.log(`
+${C.warn}Kein Ping gesetzt.${C.off} Pruefen: gibt es die Rolle ` +
+        `"${cfg.broadcaster.pingRole}" auf dem Server, und ist ein Bot-Token hinterlegt?`);
+    }
+    if (!echt) console.log(`
+${C.dim}Mit --echt geht derselbe Post in die Ankuendigungen, ` +
+      `mit Ping.${C.off}`);
+  } else {
+    console.log(`
+${C.bad}${result.status}${C.off}: ${result.reason}`);
+    process.exitCode = 1;
+  }
+}
+
+/** OBS kurz verbinden, damit ein Screenshot moeglich ist. */
+async function withObs() {
+  const { createObs } = await import('#obs/client.ts');
+  const { readSecrets } = await import('#twitch/secrets.ts');
+  const abort = new AbortController();
+  const client = createObs(`ws://${OBS_HOST}:${OBS_PORT}`,
+    () => readSecrets().obs?.password, {});
+  void client.run(abort.signal);
+  // Kurz auf die Identifikation warten; ohne sie gibt es keinen Screenshot,
+  // und das ist kein Grund zum Abbruch.
+  for (let i = 0; i < 30 && !client.isConnected(); i++) {
+    await new Promise((r) => setTimeout(r, 100));
+  }
+  return { client, stop: () => abort.abort() };
+}
+
 async function discordBot(): Promise<void> {
   const { createBot } = await import('#discord/bot.ts');
   const bot = createBot();
@@ -673,7 +844,10 @@ function help(): void {
   console.log(`
 abdulive  -  Werkzeugkasten fuer Abdus Stream
 
-  ingest                       Daemon starten (Chat, Zuschauerzahlen, Einrichtungsseite)
+  hub                          Zentraler Dienst: Chat, Discord, Go-Live, Overlay
+  ingest                       Daemon auf dem Stream-PC (OBS, !clip, Rendern)
+  overlay [--demo]             Nur den Overlay-Server, --demo mit erfundenem Betrieb
+  web:passwort [wort]          Passwort fuer die Weboberflaeche setzen
   check-env                    Umgebung pruefen
   db:migrate                   Migrationen anwenden
   fetch-model [name]           Whisper-Modell laden (Vorgabe: ggml-small.bin)
@@ -684,6 +858,8 @@ abdulive  -  Werkzeugkasten fuer Abdus Stream
   discord:setup [--server ID] [--vorschau]
                                Discord-Server nach docs/projects/discord-community.md einrichten
   discord:bot                  Rollen-Knoepfe bedienen (laeuft sonst im Daemon mit)
+  golive:test [--echt]         Go-Live-Post proben: ohne Argument in den internen
+                               Kanal ohne Ping, mit --echt in die Ankuendigungen
   help                         Diese Uebersicht
 
 Einrichtung nach dem Start: http://localhost:4792/setup
@@ -715,6 +891,14 @@ try {
     }
     case 'render': await renderOne(args[0], args[1]); break;
     case 'discord:setup': await discordSetup(args); break;
+    case 'golive:test': await goLiveTest(args); break;
+    case 'hub': {
+      const { runHub } = await import('../hub/src/index.ts');
+      await runHub();
+      break;
+    }
+    case 'overlay': await overlayOnly(args); break;
+    case 'web:passwort': await webPasswort(args); break;
     case 'discord:bot': await discordBot(); break;
     case 'help': case undefined: help(); break;
     default:

apps/hub/src/index.ts

@@ -0,0 +1,481 @@
+// Der zentrale Dienst. Laeuft in einem LXC, nicht auf Abdus Rechner.
+//
+// Was hier drin ist, braucht nur Internet: Twitch-Chat, Twitch-API, Discord
+// und die Overlay-Seiten. Das laeuft damit auch dann, wenn der Stream-PC aus
+// ist - der Rollen-Knopf auf Discord antwortet nachts um drei, und die
+// Go-Live-Ankuendigung haengt nicht daran, dass jemand den Daemon gestartet
+// hat.
+//
+// Was NICHT hier drin ist und auch nicht hierher kann: OBS, der
+// Wiedergabepuffer, ffmpeg und whisper. Die haengen an den Dateien auf dem
+// Stream-PC und bleiben in `abdulive ingest`.
+//
+//   Hub (LXC)                          Stream-PC (abdulive ingest)
+//   ---------                          --------------------------
+//   Chat lesen und senden              OBS-Wiedergabepuffer
+//   Zuschauerzahlen, Sitzungen         !clip, Rendern, Untertitel
+//   Discord: Knopf und Go-Live
+//   Overlay-Seiten fuer OBS
+//
+// Beide schreiben in ihre eigene Datenbank. Das ist bewusst: eine gemeinsame
+// waere eine Netzwerkfreigabe fuer SQLite, und das ist der zuverlaessigste
+// Weg zu einer beschaedigten Datei.
+
+import { loadConfig } 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 { createBot } from '#discord/bot.ts';
+import { helix } from '#twitch/helix.ts';
+import { readSecrets } from '#twitch/secrets.ts';
+import { createEventSub } from '#twitch/eventsub.ts';
+import { createIrc, type UserNotice } from '#twitch/irc.ts';
+import { streamWechsel } from '#twitch/stream-state.ts';
+import {
+  ausEventSub, createDrossel, createWarteschlange, entscheiden,
+} from './rewards.ts';
+import { createAnnouncer, type StreamInfo } from '../../broadcaster/src/golive.ts';
+import { createSpikeDetector } from '../../ingest/src/spike.ts';
+import { createBus, type AlertEvent, type Bus } from '../../overlay-server/src/bus.ts';
+import { startOverlayServer } from '../../overlay-server/src/server.ts';
+import { ulid } from '#shared/events.ts';
+
+const log = logger('hub');
+
+export async function runHub(): Promise<void> {
+  const cfg = loadConfig();
+  const db = openDb();
+  const repo = repos(db);
+  const spikes = createSpikeDetector(cfg.ingest.spike);
+  const bus = createBus();
+
+  const state = {
+    startedAt: new Date().toISOString(),
+    streamId: null as string | null,
+    broadcasterId: null as string | null,
+    lastSeenLive: 0,
+    viewers: 0,
+    title: '',
+    category: '',
+    startedLiveAt: null as string | null,
+    chatConnected: false,
+    chatAnonymous: true,
+    messages: 0,
+    spikeCandidates: 0,
+    eventsub: false,
+    lastError: null as string | null,
+  };
+
+  const abort = new AbortController();
+  const stop = (signal: string): void => {
+    log.info(`${signal} empfangen, fahre herunter`);
+    abort.abort();
+    setTimeout(() => { closeDb(); process.exit(0); }, 500);
+  };
+  process.on('SIGINT', () => stop('SIGINT'));
+  process.on('SIGTERM', () => stop('SIGTERM'));
+
+  // ---- Go-Live-Ankuendigung. Ohne OBS gibt es kein eigenes Vorschaubild;
+  // dann nimmt der Post Twitchs Vorschau, und die ist meist gut genug.
+  const announcer = createAnnouncer({
+    cfg: {
+      enabled: cfg.broadcaster.enabled,
+      mode: cfg.broadcaster.mode,
+      delaySeconds: cfg.broadcaster.delaySecondsAfterOnline,
+      cooldownHours: cfg.broadcaster.cooldownHours,
+      pingRole: cfg.broadcaster.pingRole,
+      obsScreenshot: false,
+    },
+    repo,
+  });
+
+  // ---- Kanalpunkte
+  const warteschlange = createWarteschlange();
+  const drossel = createDrossel();
+
+  /** Punkte zurueckgeben. Ohne den Scope nicht moeglich - dann nur melden. */
+  const erstatten = async (rewardId: string, redemptionId: string): Promise<void> => {
+    if (!state.broadcasterId) return;
+    if (!helix.hasScope('channel:manage:redemptions')) {
+      log.warn('Punkte nicht erstattet: dem verbundenen Konto fehlt ' +
+        'channel:manage:redemptions. Einmal neu verbinden unter /setup.');
+      return;
+    }
+    try {
+      await helix.setRedemptionStatus(state.broadcasterId, rewardId, redemptionId, 'CANCELED');
+    } catch (err) {
+      log.warn(`Punkte nicht erstattet: ${(err as Error).message}`);
+    }
+  };
+
+  const vorlesen = (id: string, name: string, text: string): void => {
+    bus.publish({ kind: 'tts', id, name, text });
+    log.info(`Wird vorgelesen (${name}): ${text.slice(0, 60)}`);
+  };
+
+  const behandleEinloesung = (roh: Record<string, unknown>): void => {
+    const e = ausEventSub(roh);
+    if (!e) { log.warn('Einloesung ohne verwertbare Felder - ignoriert'); return; }
+
+    repo.event('points.redeemed', {
+      reward: e.rewardTitle, kosten: e.kosten, login: e.userName, text: e.text,
+    }, state.streamId, e.userId, e.ts);
+
+    const ent = entscheiden(e, cfg.rewards, drossel.zaehlen(e.userId));
+    drossel.merken(e.userId);
+
+    switch (ent.art) {
+      case 'unbekannt':
+        log.debug(`Belohnung "${e.rewardTitle}" hat keine hinterlegte Aktion`);
+        break;
+
+      case 'abgelehnt':
+        log.info(`Einloesung von ${e.userName} abgelehnt: ${ent.grund}`);
+        void erstatten(e.rewardId, e.id);
+        break;
+
+      case 'ausfuehren':
+        // Immer dieselbe Einblendung: sie sagt, wer was eingeloest hat. Die
+        // Aktion entscheidet nur, was zusaetzlich passiert. Frueher stand
+        // hier die Abo-Einblendung - eine Belohnung als "Neues Abo" ins Bild
+        // zu schreiben ist eine Falschmeldung, die in der Aufzeichnung bleibt.
+        bus.publish({
+          kind: 'alert', type: 'reward', id: ulid(), name: e.userName,
+          reward: e.rewardTitle,
+          ...(e.kosten ? { amount: e.kosten } : {}),
+          ...(e.text ? { message: e.text } : {}),
+        });
+        if (ent.aktion === 'konfetti') bus.publish({ kind: 'konfetti', staerke: 1 });
+        if (ent.aktion === 'emote-regen') bus.publish({ kind: 'hype', score: 5, rate: 40 });
+        log.info(`${e.userName} hat "${e.rewardTitle}" eingeloest -> ${ent.aktion}`);
+        break;
+
+      case 'vorlesen':
+        vorlesen(ulid(), e.userName, ent.text);
+        break;
+
+      case 'warteschlange':
+        warteschlange.einreihen({
+          id: ulid(), name: e.userName, text: ent.text,
+          rewardId: e.rewardId, redemptionId: e.id, seit: new Date().toISOString(),
+        });
+        log.info(`${e.userName} wartet auf Freigabe zum Vorlesen`);
+        break;
+    }
+  };
+
+  // ---- EventSub. Nur sinnvoll mit verbundenem Konto: der WebSocket-Weg
+  // verlangt fuer das Abonnement dasselbe Nutzer-Token.
+  const eventsub = createEventSub({
+    onSession: async (sessionId) => {
+      const id = state.broadcasterId || (await helix.userByLogin(cfg.channel.login))?.id;
+      if (!id) throw new Error('Broadcaster-ID unbekannt');
+      state.broadcasterId = id;
+      await helix.subscribe({
+        type: 'channel.channel_points_custom_reward_redemption.add',
+        version: '1',
+        condition: { broadcaster_user_id: id },
+      }, sessionId);
+      log.info('Abonniert: Kanalpunkt-Einloesungen');
+    },
+    onNotification: (typ, event) => {
+      if (typ === 'channel.channel_points_custom_reward_redemption.add') {
+        behandleEinloesung(event);
+      }
+    },
+    onRevocation: (typ) => log.error(
+      `Twitch hat "${typ}" entzogen - meist zurueckgezogene Freigabe. Neu verbinden unter /setup.`),
+    onConnected: () => { state.eventsub = true; },
+    onClosed: () => { state.eventsub = false; },
+  });
+
+  // ---- Discord (Rollen-Knoepfe)
+  const bot = createBot();
+  if (bot) void bot.run(abort.signal);
+
+  // ---- Chat
+  const irc = createIrc(cfg.channel.login, {
+    onConnected: (anonymous) => {
+      state.chatConnected = true;
+      state.chatAnonymous = anonymous;
+    },
+    onMessage: (msg) => {
+      try {
+        repo.recordMessage(msg, state.streamId);
+        state.messages++;
+        repo.event('chat.message', {
+          text: msg.text, login: msg.login, isFirst: msg.isFirst, bits: msg.bits,
+        }, state.streamId, msg.userId, msg.ts);
+
+        bus.publish({
+          kind: 'chat',
+          id: msg.id,
+          userId: msg.userId,
+          login: msg.login,
+          name: msg.displayName,
+          text: msg.text,
+          color: msg.color,
+          emotes: msg.emotes,
+          gifs: msg.gifs,
+          first: msg.isFirst,
+          mod: msg.isMod,
+          sub: msg.isSub,
+          broadcaster: msg.isBroadcaster,
+          bits: msg.bits,
+        });
+
+        if (msg.isFirst) {
+          repo.event('chat.first_message', { login: msg.login }, state.streamId, msg.userId, msg.ts);
+          bus.publish({ kind: 'alert', type: 'first', id: ulid(), name: msg.displayName });
+          log.info(`Erstchatter: ${msg.displayName}`);
+        }
+        if (msg.bits > 0) {
+          repo.event('cheer', { bits: msg.bits, login: msg.login }, state.streamId, msg.userId, msg.ts);
+          bus.publish({
+            kind: 'alert', type: 'cheer', id: ulid(),
+            name: msg.displayName, amount: msg.bits, message: msg.text,
+          });
+        }
+
+        const spike = spikes.push(msg);
+        if (spike) {
+          state.spikeCandidates++;
+          repo.event('chat.spike', spike, state.streamId, null, msg.ts);
+          bus.publish({ kind: 'hype', score: spike.score, rate: spike.rate });
+          log.info(`Spitze im Chat: Punktzahl ${spike.score} (${spike.rate} Nachrichten ` +
+            `von ${spike.chatters} Leuten, Basislinie ${spike.baseline})`);
+        }
+      } catch (err) {
+        state.lastError = (err as Error).message;
+        log.error(`Nachricht nicht verarbeitet: ${(err as Error).message}`);
+      }
+    },
+    onUserNotice: (notice) => {
+      const alert = toAlert(notice);
+      if (!alert) return;
+      repo.event(eventTypeFor(notice), notice, state.streamId);
+      bus.publish(alert);
+      log.info(notice.systemMessage || `${notice.kind} von ${notice.displayName}`);
+    },
+    onClear: (what) => bus.publish({ kind: 'clear', ...what }),
+  });
+  void irc.run(abort.signal);
+
+  // ---- Die eine oeffentliche Tuer
+  const server = startOverlayServer({
+    port: cfg.overlay.port,
+    channel: cfg.channel.login,
+    publicUrl: cfg.publicUrl,
+    bus,
+    setup: {
+      channel: cfg.channel.login,
+      port: cfg.overlay.port,
+      health: () => ({
+        ...state,
+        rolle: 'hub',
+        db: repo.stats(),
+        discordConnected: bot ? bot.isConnected() : false,
+        goLive: announcer.lastResult() ?? 'noch nichts',
+        overlayClients: bus.clients(),
+        ttsWartend: warteschlange.liste().length,
+        spikeArmed: spikes.isArmed(),
+        note: state.chatAnonymous
+          ? 'Chat wird anonym mitgelesen. Zum Senden und fuer Helix: Einrichtung unter /setup.'
+          : undefined,
+      }),
+    },
+    tts: {
+      wartend: () => warteschlange.liste()
+        .map((w) => ({ id: w.id, name: w.name, text: w.text, seit: w.seit })),
+      freigeben: (id) => {
+        const w = warteschlange.entnehmen(id);
+        if (!w) return false;
+        vorlesen(w.id, w.name, w.text);
+        return true;
+      },
+      ablehnen: (id) => {
+        const w = warteschlange.entnehmen(id);
+        if (!w) return false;
+        log.info(`Vorlesen abgelehnt (${w.name})`);
+        void erstatten(w.rewardId, w.redemptionId);
+        return true;
+      },
+      alleAblehnen: () => {
+        const alle = warteschlange.leeren();
+        for (const w of alle) void erstatten(w.rewardId, w.redemptionId);
+        if (alle.length) log.info(`${alle.length} wartende Nachricht(en) abgelehnt`);
+        return alle.length;
+      },
+      still: () => {
+        bus.publish({ kind: 'still' });
+        const alle = warteschlange.leeren();
+        for (const w of alle) void erstatten(w.rewardId, w.redemptionId);
+        log.warn('Panikschalter: Vorlesen gestoppt und Warteschlange geleert');
+      },
+    },
+  });
+  void server;
+
+  // ---- Zuschauerzahlen und Stream-Sitzungen
+  void pollStream(cfg, repo, state, bus, abort.signal, (stream) => announcer.announce(stream));
+
+  // ---- EventSub erst, wenn ein Konto verbunden ist. Ohne Nutzer-Token
+  // wuerde der Verbindungsversuch im Minutentakt scheitern und nichts als
+  // Protokollrauschen erzeugen.
+  if (cfg.rewards.enabled && readSecrets().tokens.broadcaster) {
+    void eventsub.run(abort.signal);
+  } else if (cfg.rewards.enabled) {
+    log.warn('Kanalpunkt-Aktionen sind an, aber kein Konto verbunden - ' +
+      'EventSub bleibt aus. Verbinden unter /setup.');
+  }
+
+  log.info(`Hub laeuft fuer #${cfg.channel.login}`);
+  await new Promise<void>((resolve) => abort.signal.addEventListener('abort', () => resolve()));
+}
+
+// ---------------------------------------------------------------- Ereignisse
+
+/** USERNOTICE in eine Einblendung uebersetzen. Unbekanntes faellt weg. */
+export function toAlert(n: UserNotice): AlertEvent | null {
+  const gemeinsam = {
+    kind: 'alert' as const,
+    id: ulid(),
+    name: n.displayName,
+    ...(n.months !== undefined ? { months: n.months } : {}),
+    ...(n.tier ? { tier: n.tier } : {}),
+    ...(n.text ? { message: n.text } : {}),
+  };
+  switch (n.kind) {
+    case 'sub': case 'resub':
+      return { ...gemeinsam, type: n.kind };
+    case 'subgift':
+      return { ...gemeinsam, type: 'subgift', ...(n.recipient ? { recipient: n.recipient } : {}) };
+    case 'massgift':
+      return { ...gemeinsam, type: 'massgift', ...(n.count ? { amount: n.count } : {}) };
+    case 'raid':
+      return { ...gemeinsam, type: 'raid', ...(n.count ? { amount: n.count } : {}) };
+    default:
+      // Ankuendigungen und alles Unbekannte bekommen keine Einblendung. Eine
+      // Einblendung fuer etwas, das man nicht kennt, sieht kaputt aus.
+      return null;
+  }
+}
+
+function eventTypeFor(n: UserNotice): 'subscribe' | 'subgift' | 'raid.in' | 'chat.message' {
+  if (n.kind === 'subgift' || n.kind === 'massgift') return 'subgift';
+  if (n.kind === 'raid') return 'raid.in';
+  if (n.kind === 'sub' || n.kind === 'resub') return 'subscribe';
+  return 'chat.message';
+}
+
+// ---------------------------------------------------------------- Abfrage
+
+type State = {
+  streamId: string | null; broadcasterId: string | null; lastSeenLive: number;
+  viewers: number; title: string; category: string; startedLiveAt: string | null;
+  lastError: string | null;
+};
+
+async function pollStream(
+  cfg: ReturnType<typeof loadConfig>,
+  repo: ReturnType<typeof repos>,
+  state: State,
+  bus: Bus,
+  signal: AbortSignal,
+  onOnline: (stream: StreamInfo) => void
+): Promise<void> {
+  const gapMs = cfg.ingest.sessionGapMinutes * 60_000;
+  let warnedMissingAuth = false;
+
+  const sendState = (live: boolean): void => {
+    bus.publish({
+      kind: 'stream', live,
+      title: state.title, category: state.category,
+      viewers: state.viewers, startedAt: state.startedLiveAt,
+    });
+  };
+
+  // Ein laufender Stream wird beim Start uebernommen, damit ein Neustart
+  // mitten im Stream keine zweite Sitzung anlegt - und keinen zweiten Ping.
+  const open = repo.latestOpenStream();
+  if (open) {
+    state.streamId = open.id;
+    state.startedLiveAt = open.started_at;
+    state.lastSeenLive = Date.now();
+    log.info(`Laufende Sitzung uebernommen: ${open.id} seit ${open.started_at}`);
+  }
+
+  while (!signal.aborted) {
+    try {
+      // Client-ID und Secret genuegen: `GET /users` und `GET /streams` laufen
+      // mit einem App-Token. Auf das verbundene Konto wartet hier niemand.
+      if (!helix.canReadStream()) {
+        if (!warnedMissingAuth) {
+          log.warn('Keine Twitch-App hinterlegt - Zuschauerzahlen, Sitzungen und die ' +
+            'Go-Live-Ankuendigung bleiben aus. Einrichtung unter /setup.');
+          warnedMissingAuth = true;
+        }
+      } else {
+        warnedMissingAuth = false;
+        const userId = state.broadcasterId || cfg.channel.userId ||
+          (await helix.userByLogin(cfg.channel.login))?.id || '';
+        if (userId) state.broadcasterId = userId;
+        const live = userId ? await helix.streamByUserId(userId) : null;
+        const now = Date.now();
+
+        const wechsel = streamWechsel({
+          live: live ? { id: live.id, startedAt: live.started_at } : null,
+          laufendeId: state.streamId,
+          zuletztLive: state.lastSeenLive,
+          jetzt: now,
+          luckeMs: gapMs,
+        });
+
+        if (wechsel.art === 'gestartet') {
+          state.streamId = wechsel.id;
+          repo.openStream(wechsel.id, wechsel.startedAt);
+          repo.event('stream.online',
+            { title: live!.title, category: live!.game_name }, wechsel.id);
+          log.info(`Stream online: "${live!.title}" (${live!.game_name}), ` +
+            `gestartet ${wechsel.startedAt}`);
+          onOnline({
+            id: wechsel.id,
+            title: live!.title,
+            category: live!.game_name,
+            login: cfg.channel.login,
+            startedAt: wechsel.startedAt,
+            thumbnailUrl: live!.thumbnail_url,
+          });
+        } else if (wechsel.art === 'fortsetzung') {
+          log.info(`Neue Twitch-Stream-ID ${wechsel.neueId}, behandle als Fortsetzung ` +
+            `von ${state.streamId} - keine zweite Ankuendigung`);
+        } else if (wechsel.art === 'beendet') {
+          repo.closeStream(wechsel.id, new Date().toISOString());
+          repo.event('stream.offline', {}, wechsel.id);
+          log.info(`Stream offline: ${wechsel.id}`);
+          state.streamId = null;
+          state.viewers = 0;
+          state.startedLiveAt = null;
+          sendState(false);
+        }
+
+        if (live) {
+          state.title = live.title;
+          state.category = live.game_name;
+          state.startedLiveAt = live.started_at;
+          state.lastSeenLive = now;
+          state.viewers = live.viewer_count;
+          if (state.streamId) {
+            repo.sample(state.streamId, new Date().toISOString(), live.viewer_count);
+          }
+          sendState(true);
+        }
+      }
+    } catch (err) {
+      state.lastError = (err as Error).message;
+      log.error(`Abfrage fehlgeschlagen: ${(err as Error).message}`);
+    }
+    try { await sleep(cfg.ingest.viewerPollSeconds * 1000, signal); } catch { return; }
+  }
+}

apps/ingest/src/server.ts

@@ -59,18 +59,45 @@ async function handle(
   req: http.IncomingMessage, res: http.ServerResponse, deps: ServerDeps
 ): Promise<void> {
   const url = new URL(req.url ?? '/', `http://localhost:${deps.port}`);
+  // Auf dem Stream-PC ist die Einrichtung das Einzige, was dieser Server
+  // kann - also faengt die Wurzel dorthin. Im Hub steht auf / die
+  // Overlay-Uebersicht, deshalb liegt die Weiche hier und nicht in
+  // `setupRoutes`.
+  if (url.pathname === '/') {
+    res.writeHead(303, { location: '/setup' });
+    res.end();
+    return;
+  }
+  if (await setupRoutes(req, res, deps, url)) return;
+  res.writeHead(404, { 'content-type': 'text/plain; charset=utf-8' });
+  res.end('Nicht gefunden');
+}
+
+/**
+ * Die Routen der Einrichtung, ohne 404 am Ende.
+ *
+ * Ausgelagert, weil sie seit dem zentralen Betrieb an zwei Servern haengen:
+ * hier auf dem Stream-PC und im Overlay-Server des Hubs. Der Rueckgabewert
+ * sagt, ob die Anfrage beantwortet wurde - wer sie einbindet, macht sonst
+ * selbst weiter.
+ */
+export async function setupRoutes(
+  req: http.IncomingMessage, res: http.ServerResponse, deps: ServerDeps, url: URL
+): Promise<boolean> {
   const html = (body: string, code = 200) => {
     res.writeHead(code, { 'content-type': 'text/html; charset=utf-8',
       'cache-control': 'no-store', 'referrer-policy': 'no-referrer' });
     res.end(body);
+    return true;
   };
   const back = (kind: 'ok' | 'err', text: string) => {
     res.writeHead(303, { location: `/setup?${kind}=${encodeURIComponent(text)}` });
     res.end();
+    return true;
   };
 
   // ---- Einrichtungsseite
-  if (url.pathname === '/' || url.pathname === '/setup') {
+  if (url.pathname === '/setup') {
     const ok = url.searchParams.get('ok');
     const err = url.searchParams.get('err');
     return html(setupPage(secretStatus(), {
@@ -119,7 +146,7 @@ async function handle(
     try {
       res.writeHead(303, { location: authorizeUrl(role, deps.port) });
       res.end();
-      return;
+      return true;
     } catch (err) {
       return back('err', (err as Error).message);
     }
@@ -144,9 +171,8 @@ async function handle(
   if (url.pathname === '/health') {
     res.writeHead(200, { 'content-type': 'application/json; charset=utf-8' });
     res.end(JSON.stringify(deps.health(), null, 2));
-    return;
+    return true;
   }
 
-  res.writeHead(404, { 'content-type': 'text/plain; charset=utf-8' });
-  res.end('Nicht gefunden');
+  return false;
 }