Stream-Erkennung: eine geprüfte Funktion statt zweimal ausgeschrieben

Wann ist ein Stream ein *neuer* Stream? Die Entscheidung stand doppelt in Hub
und Ingest und war ungetestet - dabei haengt genau an ihr, ob der Go-Live-Ping
zum richtigen Zeitpunkt und nur einmal rausgeht.

Jetzt rein in stream-state.ts, mit zehn Tests. Darunter die Faelle, die im
Betrieb selten sind und dann mitten im Stream zuschlagen:

- Neue Twitch-ID nach kurzem Aussetzer ist eine Fortsetzung, kein zweiter
  Ping. Twitch vergibt bei einem Verbindungsabbruch eine neue ID.
- Eine einzelne leere Antwort beendet den Stream nicht. Wer daraufhin die
  Sitzung schliesst, hat beim naechsten Abruf einen "neuen" Stream.
- Ein Durchlauf ueber 90 Minuten mit zwei Aussetzern, im Minutentakt
  abgefragt, ergibt genau einen Start und genau ein Ende.

Dazu App-Token fuer Helix: `GET /users` und `GET /streams` kommen ohne
Nutzer-Token aus. Das ist der Unterschied zwischen "laeuft" und "wartet auf
einen Klick" - ohne diese Unterscheidung stand der ganze Dienst still, bis
der Broadcaster sein Konto verbindet. Endpunkte, die etwas tun oder Privates
lesen, verlangen weiterhin ein Nutzer-Token, und `appOk` sagt je Aufruf, was
gilt.

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

apps/ingest/src/index.ts

@@ -16,6 +16,8 @@ import { isWindows } from '#core/platform.ts';
 import { helix } from '#twitch/helix.ts';
 import { createIrc } from '#twitch/irc.ts';
 import { readSecrets } from '#twitch/secrets.ts';
+import { streamWechsel } from '#twitch/stream-state.ts';
+import { createAnnouncer, type StreamInfo } from '../../broadcaster/src/golive.ts';
 import { createClipper } from './clips.ts';
 import { createRenderer } from './render.ts';
 import { createSpikeDetector } from './spike.ts';
@@ -42,6 +44,7 @@ export async function runIngest(): Promise<void> {
     messages: 0,
     spikeCandidates: 0,
     clipsTriggered: 0,
+    goLive: 'noch nichts' as string,
     lastError: null as string | null,
   };
 
@@ -111,6 +114,20 @@ export async function runIngest(): Promise<void> {
     onRawReady: (job) => renderer.enqueue(job),
   });
 
+  // ---- Go-Live-Ankuendigung auf Discord
+  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: cfg.broadcaster.obsScreenshot,
+    },
+    repo,
+    obs,
+  });
+
   startServer({
     channel: cfg.channel.login,
     port: cfg.dashboard.port,
@@ -118,6 +135,7 @@ export async function runIngest(): Promise<void> {
       ...state,
       db: repo.stats(),
       discordConnected: bot ? bot.isConnected() : false,
+      goLive: announcer.lastResult() ?? state.goLive,
       spikeArmed: spikes.isArmed(),
       clipsWaitingForFile: clipper.pendingCount(),
       renderQueue: renderer.pending(),
@@ -177,7 +195,10 @@ export async function runIngest(): Promise<void> {
   if (bot) void bot.run(abort.signal);
 
   // ---- Zuschauerzahlen und Stream-Sessions
-  void pollStream(cfg, repo, state, abort.signal);
+  void pollStream(cfg, repo, state, abort.signal, (stream) => {
+    state.goLive = 'laeuft';
+    announcer.announce(stream);
+  });
 
   log.info(`Ingest laeuft fuer #${cfg.channel.login}`);
   await new Promise<void>((resolve) => abort.signal.addEventListener('abort', () => resolve()));
@@ -185,14 +206,16 @@ export async function runIngest(): Promise<void> {
 
 type State = {
   streamId: string | null; broadcasterId: string | null;
-  lastSeenLive: number; viewers: number; lastError: string | null;
+  lastSeenLive: number; viewers: number; goLive: string; lastError: string | null;
 };
 
 async function pollStream(
   cfg: ReturnType<typeof loadConfig>,
   repo: ReturnType<typeof repos>,
   state: State,
-  signal: AbortSignal
+  signal: AbortSignal,
+  /** Genau einmal je neu erkanntem Stream. Fortsetzungen zaehlen nicht. */
+  onOnline: (stream: StreamInfo) => void
 ): Promise<void> {
   const gapMs = cfg.ingest.sessionGapMinutes * 60_000;
   let warnedMissingAuth = false;
@@ -208,10 +231,12 @@ async function pollStream(
 
   while (!signal.aborted) {
     try {
-      if (!readSecrets().tokens.broadcaster) {
+      // Client-ID und Secret genuegen: `GET /users` und `GET /streams` laufen
+      // mit einem App-Token, auch ohne verbundenes Konto.
+      if (!helix.canReadStream()) {
         if (!warnedMissingAuth) {
-          log.warn('Kein Broadcaster-Token - Zuschauerzahlen und Stream-Sitzungen bleiben aus. ' +
-            'Einrichtung unter /setup.');
+          log.warn('Keine Twitch-App hinterlegt - Zuschauerzahlen und Stream-Sitzungen ' +
+            'bleiben aus. Einrichtung unter /setup.');
           warnedMissingAuth = true;
         }
       } else {
@@ -223,33 +248,49 @@ async function pollStream(
         const live = userId ? await helix.streamByUserId(userId) : null;
         const now = Date.now();
 
-        if (live) {
-          if (state.streamId !== live.id) {
-            // Kurze Unterbrechungen gelten als dieselbe Sitzung, sonst
-            // zerfaellt ein Stream bei jedem Aussetzer in Fragmente.
-            const wasRecent = now - state.lastSeenLive < gapMs;
-            if (state.streamId && wasRecent) {
-              log.info(`Neue Twitch-Stream-ID ${live.id}, behandle als Fortsetzung von ${state.streamId}`);
-            } else {
-              state.streamId = live.id;
-              repo.openStream(live.id, live.started_at);
+        // Die Entscheidung "neuer Stream oder Fortsetzung" steht in
+        // #twitch/stream-state.ts und ist dort geprueft. Sie haengt daran,
+        // ob ein zweiter Rundruf rausgeht.
+        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 }, live.id);
-              log.info(`Stream online: "${live.title}" (${live.game_name})`);
-            }
+            { 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}`);
+        } 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;
         }
+
+        if (live) {
           state.lastSeenLive = now;
           state.viewers = live.viewer_count;
           if (state.streamId) {
             repo.sample(state.streamId, new Date().toISOString(), live.viewer_count);
           }
-        } else if (state.streamId && now - state.lastSeenLive > gapMs) {
-          const ended = new Date().toISOString();
-          repo.closeStream(state.streamId, ended);
-          repo.event('stream.offline', {}, state.streamId);
-          log.info(`Stream offline: ${state.streamId}`);
-          state.streamId = null;
-          state.viewers = 0;
         }
       }
     } catch (err) {

packages/twitch/src/helix.ts

@@ -4,7 +4,7 @@
 
 import { logger } from '#core/logger.ts';
 import { retry, sleep } from '#core/retry.ts';
-import { accessToken } from './oauth.ts';
+import { accessToken, appToken } from './oauth.ts';
 import { readSecrets, type Role } from './secrets.ts';
 
 const log = logger('helix');
@@ -49,13 +49,33 @@ export type HelixStream = {
 
 export type HelixUser = { id: string; login: string; display_name: string };
 
-async function call<T>(role: Role, endpoint: string, init: RequestInit = {}): Promise<T> {
-  const app = readSecrets().apps[role];
+/**
+ * `appOk` heisst: dieser Endpunkt kommt auch ohne Nutzer-Token aus.
+ *
+ * Das ist kein Detail, sondern der Unterschied zwischen "laeuft" und "wartet
+ * auf einen Klick": Stream-Erkennung und damit die Go-Live-Ankuendigung
+ * brauchen nur `GET /users` und `GET /streams`, und beide sind oeffentlich.
+ * Ohne diese Unterscheidung stuende der ganze Dienst still, bis der
+ * Broadcaster sein Konto verbindet.
+ */
+type CallOpts = { appOk?: boolean };
+
+async function call<T>(
+  role: Role, endpoint: string, init: RequestInit = {}, opts: CallOpts = {}
+): Promise<T> {
+  const secrets = readSecrets();
+  const app = secrets.apps[role];
   if (!app?.clientId) throw new Error(`Fuer "${role}" ist keine Client-ID hinterlegt.`);
+  const hatNutzerToken = Boolean(secrets.tokens[role]);
+  if (!hatNutzerToken && !opts.appOk) {
+    throw new Error(`"${role}" ist nicht verbunden - Einrichtung unter /setup.`);
+  }
 
   return await retry(async () => {
     await bucket.take();
-    const token = await accessToken(role);
+    // Das Nutzer-Token gewinnt, wenn es da ist: es sieht mehr (Abos, Chatter)
+    // und zaehlt auf ein eigenes Kontingent.
+    const token = hatNutzerToken ? await accessToken(role) : await appToken(role);
     const res = await fetch(`${BASE}${endpoint}`, {
       ...init,
       headers: {
@@ -85,15 +105,26 @@ async function call<T>(role: Role, endpoint: string, init: RequestInit = {}): Pr
 
 export const helix = {
   async userByLogin(login: string): Promise<HelixUser | null> {
-    const r = await call<{ data: HelixUser[] }>('broadcaster', `/users?login=${encodeURIComponent(login)}`);
+    const r = await call<{ data: HelixUser[] }>(
+      'broadcaster', `/users?login=${encodeURIComponent(login)}`, {}, { appOk: true });
     return r.data[0] ?? null;
   },
 
   async streamByUserId(userId: string): Promise<HelixStream | null> {
-    const r = await call<{ data: HelixStream[] }>('broadcaster', `/streams?user_id=${userId}`);
+    const r = await call<{ data: HelixStream[] }>(
+      'broadcaster', `/streams?user_id=${userId}`, {}, { appOk: true });
     return r.data[0] ?? null;
   },
 
+  /**
+   * Reicht das Hinterlegte, um den Stream-Zustand abzufragen? Client-ID und
+   * Secret genuegen dafuer - ein verbundenes Konto ist nicht noetig.
+   */
+  canReadStream(): boolean {
+    const app = readSecrets().apps.broadcaster;
+    return Boolean(app?.clientId && app.clientSecret);
+  },
+
   async chatters(broadcasterId: string, moderatorId: string): Promise<string[]> {
     const out: string[] = [];
     let cursor = '';
@@ -108,6 +139,48 @@ export const helix = {
     return out;
   },
 
+  /**
+   * Abonnement fuer die WebSocket-Sitzung anlegen.
+   *
+   * Zwingend mit dem Nutzer-Token: Twitch verlangt fuer den WebSocket-Weg
+   * dasselbe Token, mit dem die Sitzung geoeffnet wurde. Ein App-Token wird
+   * hier abgelehnt, und die Fehlermeldung sagt nicht, warum.
+   */
+  async subscribe(
+    sub: { type: string; version: string; condition: Record<string, string> },
+    sessionId: string,
+  ): Promise<{ id: string }> {
+    const r = await call<{ data: Array<{ id: string }> }>('broadcaster', '/eventsub/subscriptions', {
+      method: 'POST',
+      body: JSON.stringify({ ...sub, transport: { method: 'websocket', session_id: sessionId } }),
+    });
+    const angelegt = r.data[0];
+    if (!angelegt) throw new Error(`Abonnement ${sub.type} wurde nicht angelegt`);
+    return angelegt;
+  },
+
+  /**
+   * Eine Einloesung abschliessen oder die Punkte zurueckgeben.
+   *
+   * Braucht `channel:manage:redemptions`; `channel:read:redemptions` allein
+   * genuegt nicht. Fehlt der Scope, meldet Twitch 401 - der Aufrufer faengt
+   * das ab und macht weiter, statt die Aktion daran scheitern zu lassen.
+   */
+  async setRedemptionStatus(
+    broadcasterId: string, rewardId: string, redemptionId: string,
+    status: 'FULFILLED' | 'CANCELED',
+  ): Promise<void> {
+    await call('broadcaster',
+      `/channel_points/custom_rewards/redemptions?broadcaster_id=${broadcasterId}` +
+      `&reward_id=${encodeURIComponent(rewardId)}&id=${encodeURIComponent(redemptionId)}`,
+      { method: 'PATCH', body: JSON.stringify({ status }) });
+  },
+
+  /** Hat das verbundene Konto diesen Scope freigegeben? */
+  hasScope(scope: string): boolean {
+    return (readSecrets().tokens.broadcaster?.scopes ?? []).includes(scope);
+  },
+
   async createClip(broadcasterId: string): Promise<{ id: string; edit_url: string }> {
     const r = await call<{ data: Array<{ id: string; edit_url: string }> }>(
       'broadcaster', `/clips?broadcaster_id=${broadcasterId}`, { method: 'POST' });

packages/twitch/src/stream-state.test.ts

@@ -0,0 +1,119 @@
+// Diese Funktion entscheidet, wann der Go-Live-Ping rausgeht. Ein Fehler hier
+// ist entweder ein verpasster Stream oder ein zweiter Rundruf an Leute, die
+// sich daraufhin die Rolle abnehmen. Beides merkt man erst Tage spaeter.
+
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { streamWechsel, type StreamBeobachtung } from './stream-state.ts';
+
+const LUECKE = 5 * 60_000;                       // fuenf Minuten, wie in der Vorgabe
+const T = Date.parse('2026-09-17T20:00:00Z');
+
+function lage(over: Partial<StreamBeobachtung> = {}): StreamBeobachtung {
+  return {
+    live: null, laufendeId: null, zuletztLive: 0, jetzt: T, luckeMs: LUECKE, ...over,
+  };
+}
+
+test('kein Stream, nichts laeuft: nichts passiert', () => {
+  assert.deepEqual(streamWechsel(lage()), { art: 'unveraendert' });
+});
+
+test('der erste Stream wird als Start erkannt', () => {
+  const w = streamWechsel(lage({ live: { id: '48212345', startedAt: '2026-09-17T19:58:00Z' } }));
+  assert.deepEqual(w, { art: 'gestartet', id: '48212345', startedAt: '2026-09-17T19:58:00Z' });
+});
+
+test('derselbe Stream in der naechsten Abfrage loest nichts aus', () => {
+  // Bei 60 Sekunden Abfrageabstand passiert das den ganzen Stream lang.
+  // Wuerde es hier "gestartet" heissen, kaeme jede Minute ein Ping.
+  const w = streamWechsel(lage({
+    live: { id: '48212345', startedAt: '2026-09-17T19:58:00Z' },
+    laufendeId: '48212345', zuletztLive: T - 60_000,
+  }));
+  assert.deepEqual(w, { art: 'unveraendert' });
+});
+
+test('neue Twitch-ID nach kurzem Aussetzer ist eine Fortsetzung', () => {
+  // Der Fall aus dem Betrieb: das Internet hakt zwei Minuten, Twitch vergibt
+  // eine neue Stream-ID. Derselbe Stream, kein zweiter Ping.
+  const w = streamWechsel(lage({
+    live: { id: '48299999', startedAt: '2026-09-17T20:02:00Z' },
+    laufendeId: '48212345', zuletztLive: T - 2 * 60_000,
+  }));
+  assert.deepEqual(w, { art: 'fortsetzung', neueId: '48299999' });
+});
+
+test('neue ID nach langer Pause ist ein neuer Stream', () => {
+  const w = streamWechsel(lage({
+    live: { id: '48299999', startedAt: '2026-09-17T20:00:00Z' },
+    laufendeId: '48212345', zuletztLive: T - 6 * 60_000,
+  }));
+  assert.equal(w.art, 'gestartet');
+});
+
+test('genau an der Grenze gilt es noch als Fortsetzung', () => {
+  // Die Grenze liegt bei "kuerzer als die Luecke". Wer sie auf <= verschiebt,
+  // baut sich einen Fall, der mal so und mal so ausgeht.
+  const knappDrunter = streamWechsel(lage({
+    live: { id: 'neu', startedAt: 'x' }, laufendeId: 'alt', zuletztLive: T - (LUECKE - 1),
+  }));
+  assert.equal(knappDrunter.art, 'fortsetzung');
+
+  const knappDrueber = streamWechsel(lage({
+    live: { id: 'neu', startedAt: 'x' }, laufendeId: 'alt', zuletztLive: T - (LUECKE + 1),
+  }));
+  assert.equal(knappDrueber.art, 'gestartet');
+});
+
+test('ein einzelner fehlgeschlagener Abruf beendet den Stream nicht', () => {
+  // Twitch antwortet gelegentlich mit einer leeren Liste, obwohl der Stream
+  // laeuft. Wer daraufhin die Sitzung schliesst, bekommt beim naechsten
+  // Abruf einen "neuen" Stream - und einen zweiten Ping.
+  const w = streamWechsel(lage({ laufendeId: '48212345', zuletztLive: T - 60_000 }));
+  assert.deepEqual(w, { art: 'unveraendert' });
+});
+
+test('nach der Luecke ohne Meldung gilt der Stream als beendet', () => {
+  const w = streamWechsel(lage({ laufendeId: '48212345', zuletztLive: T - 6 * 60_000 }));
+  assert.deepEqual(w, { art: 'beendet', id: '48212345' });
+});
+
+test('ein beendeter Stream wird nicht zweimal beendet', () => {
+  const nachEnde = streamWechsel(lage({ laufendeId: null, zuletztLive: T - 60 * 60_000 }));
+  assert.deepEqual(nachEnde, { art: 'unveraendert' });
+});
+
+test('der Ablauf eines ganzen Streams ergibt genau einen Start und ein Ende', () => {
+  // Der Durchlauf, der zaehlt: 90 Minuten Stream, zwei Aussetzer, alle 60 s
+  // eine Abfrage. Am Ende muss genau ein Ping stehen.
+  let laufendeId: string | null = null;
+  let zuletztLive = 0;
+  const starts: string[] = [];
+  const enden: string[] = [];
+
+  // Minute -> was Twitch meldet. null = Aussetzer.
+  const verlauf: Array<string | null> = [
+    ...Array(20).fill('A'),        // 20 min normal
+    null, null,                    // 2 min Aussetzer
+    ...Array(30).fill('B'),        // Twitch vergibt eine neue ID
+    null,                          // kurzer Hakler
+    ...Array(20).fill('B'),
+    ...Array(10).fill(null),       // Stream zu Ende
+  ];
+
+  verlauf.forEach((id, minute) => {
+    const jetzt = T + minute * 60_000;
+    const w = streamWechsel({
+      live: id ? { id, startedAt: '2026-09-17T20:00:00Z' } : null,
+      laufendeId, zuletztLive, jetzt, luckeMs: LUECKE,
+    });
+    if (id) zuletztLive = jetzt;
+    if (w.art === 'gestartet') { laufendeId = w.id; starts.push(w.id); }
+    if (w.art === 'fortsetzung') { /* ID bewusst behalten */ }
+    if (w.art === 'beendet') { enden.push(w.id); laufendeId = null; }
+  });
+
+  assert.deepEqual(starts, ['A'], 'genau ein Ping, nicht drei');
+  assert.deepEqual(enden, ['A'], 'genau ein Streamende');
+});

packages/twitch/src/stream-state.ts

@@ -0,0 +1,58 @@
+// Wann ist ein Stream ein *neuer* Stream?
+//
+// Das klingt trivial und ist es nicht. Twitch vergibt bei einem
+// Verbindungsabbruch eine neue Stream-ID, auch wenn der Stream aus Sicht aller
+// Beteiligten derselbe ist. Wer das fuer einen Neustart haelt, legt eine
+// zweite Sitzung an und schickt einen zweiten Rundruf an alle, die die Rolle
+// Live-Ping tragen. Wer zweimal in zehn Minuten angepingt wird, nimmt sich die
+// Rolle ab, und dann ist der Kanal fuer diese Person fuer immer stumm.
+//
+// Deshalb steht die Entscheidung hier, als reine Funktion, mit Tests - statt
+// wie vorher zweimal ausgeschrieben in Hub und Ingest, wo sie auseinander
+// laufen kann, ohne dass es jemandem auffaellt.
+
+export type StreamBeobachtung = {
+  /** Was Twitch gerade meldet. `null` heisst: gerade kein Stream. */
+  live: { id: string; startedAt: string } | null;
+  /** Die Sitzung, die wir gerade fuehren. */
+  laufendeId: string | null;
+  /** Wann Twitch zuletzt "live" gemeldet hat (ms seit Epoche). */
+  zuletztLive: number;
+  jetzt: number;
+  /** Kuerzere Luecken gelten als derselbe Stream. */
+  luckeMs: number;
+};
+
+export type StreamWechsel =
+  /** Twitch meldet dieselbe Sitzung wie bisher. Nichts zu tun. */
+  | { art: 'unveraendert' }
+  /** Ein neuer Stream. Nur hier wird angekuendigt. */
+  | { art: 'gestartet'; id: string; startedAt: string }
+  /**
+   * Neue Twitch-ID, aber die Unterbrechung war kurz genug. Die Sitzung laeuft
+   * unter der bisherigen ID weiter - und es wird nicht erneut angekuendigt.
+   */
+  | { art: 'fortsetzung'; neueId: string }
+  /** Lange genug nichts mehr gemeldet: die Sitzung ist zu Ende. */
+  | { art: 'beendet'; id: string };
+
+export function streamWechsel(b: StreamBeobachtung): StreamWechsel {
+  if (b.live) {
+    if (b.laufendeId === b.live.id) return { art: 'unveraendert' };
+
+    // Es lief schon etwas, und die Unterbrechung war kurz: Fortsetzung.
+    // Ohne diesen Zweig zerfaellt ein Stream mit zwei Aussetzern in drei
+    // Sitzungen - und erzeugt drei Ankuendigungen.
+    if (b.laufendeId !== null && b.jetzt - b.zuletztLive < b.luckeMs) {
+      return { art: 'fortsetzung', neueId: b.live.id };
+    }
+    return { art: 'gestartet', id: b.live.id, startedAt: b.live.startedAt };
+  }
+
+  // Kein Stream gemeldet. Erst nach Ablauf der Luecke gilt er als beendet -
+  // sonst beendet ein einzelner fehlgeschlagener Abruf die Sitzung.
+  if (b.laufendeId !== null && b.jetzt - b.zuletztLive > b.luckeMs) {
+    return { art: 'beendet', id: b.laufendeId };
+  }
+  return { art: 'unveraendert' };
+}