Kanalpunkte: EventSub, Aktionen und Vorlesen mit Freigabe

EventSub ueber WebSocket wurde in ADR-0003 beschlossen und nie gebaut. Jetzt
gebaut, weil Kanalpunkt-Einloesungen ueber IRC nur ankommen, wenn die
Belohnung eine Texteingabe verlangt - und selbst dann ohne Einloese-ID. Ohne
die ID lassen sich Punkte nicht zurueckgeben, wenn eine Aktion scheitert.

Zwei Eigenheiten des Protokolls, die im Code stehen:

- Ein Abonnement muss innerhalb von 10 Sekunden nach dem Willkommensgruss
  stehen, sonst schliesst Twitch wieder.
- Bei `session_reconnect` wird erst die neue Verbindung aufgebaut und dann die
  alte geschlossen. Andersherum faellt genau dazwischen eine Einloesung unter
  den Tisch.

Belohnungen werden ueber ihren Titel zugeordnet, wahlweise ueber die stabile
ID; liegt beides vor, gewinnt die ID. Eine unbekannte Belohnung ist kein
Fehler - es darf welche geben, die das Toolkit nichts angehen.

Zum Vorlesen, und warum es eine Warteschlange hat: chat-control.md schliesst
TTS fuer Ausbaustufe 1 aus, und der Grund gilt. Kanalpunkte sammeln sich beim
Zugucken von allein an und sind damit kaum eine Huerde; ohne Freigabe hat man
eine Vorlesefunktion fuer Fremde im eigenen Stream. Vorgabe ist deshalb die
Freigabe von Hand, dazu Laengengrenze, Wortfilter, Drossel je Person und
Stunde und ein Panikschalter. Zu langer Text wird abgelehnt statt gekuerzt -
ein halb vorgelesener Satz ist schlimmer als gar keiner, weil niemand weiss,
ob der Rest Absicht war.

Gesprochen wird ueber `speechSynthesis`, im Browser eingebaut: kein
Cloud-Dienst, keine Kosten, kein Zuschauertext, der das Haus verlaesst. Ob
OBS' Browser-Quelle das kann, ist offen - CEF wird teils ohne Sprachausgabe
gebaut. Die Seite prueft es beim Start und blendet einen roten Hinweis ein,
statt still zu bleiben.

Kanalpunkte haben eine eigene Einblendung. Zwischendurch stand hier die des
Abos, womit eine eingeloeste Belohnung als "Neues Abo" im Bild gelandet waere -
eine Falschmeldung, die in der Aufzeichnung bleibt. Ein Test haelt das fest.

`rewards.enabled` ist aus, bis jemand bewusst Belohnungen eintraegt. Eine
Aktion, die niemand erwartet, ist im Stream schlimmer als gar keine.

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

apps/hub/src/rewards.test.ts

@@ -0,0 +1,186 @@
+// Hier entscheidet sich, ob fremder Text im Stream vorgelesen wird. Was hier
+// durchrutscht, hoert das Publikum - und die Aufzeichnung behaelt es.
+
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import {
+  ausEventSub, createDrossel, createWarteschlange, entscheiden, planFuer,
+  type Einloesung, type RewardsConfig,
+} from './rewards.ts';
+
+const CFG: RewardsConfig = {
+  enabled: true,
+  belohnungen: [
+    { titel: 'Konfetti', aktion: 'konfetti' },
+    { titel: 'Vorlesen', aktion: 'tts' },
+    { id: 'abc-123', aktion: 'emote-regen' },
+  ],
+  tts: { freigabe: true, maxZeichen: 200, blockliste: ['heilhitler'], proPersonProStunde: 3 },
+};
+
+function einloesung(over: Partial<Einloesung> = {}): Einloesung {
+  return {
+    id: 'r1', rewardId: 'xyz', rewardTitle: 'Vorlesen', userId: 'u1', userName: 'Kevin',
+    text: 'moin zusammen', kosten: 500, ts: '2026-09-17T20:00:00Z', ...over,
+  };
+}
+
+// ---------------------------------------------------------------- Zuordnung
+
+test('die ID gewinnt gegen den Titel', () => {
+  // Der Titel kann sich aendern, die ID nicht. Wer beides hinterlegt, meint
+  // die ID.
+  const p = planFuer({ rewardId: 'abc-123', rewardTitle: 'Konfetti' }, CFG.belohnungen);
+  assert.equal(p?.aktion, 'emote-regen');
+});
+
+test('der Titel wird ohne Ruecksicht auf Gross-/Kleinschreibung gefunden', () => {
+  assert.equal(planFuer({ rewardId: 'x', rewardTitle: 'kOnFeTtI' }, CFG.belohnungen)?.aktion,
+    'konfetti');
+});
+
+test('eine unbekannte Belohnung ist kein Fehler', () => {
+  // Abdu darf Belohnungen anlegen, die mit dem Toolkit nichts zu tun haben.
+  const e = entscheiden(einloesung({ rewardTitle: 'Nachricht anheften' }), CFG);
+  assert.equal(e.art, 'unbekannt');
+});
+
+// ---------------------------------------------------------------- Einfache Aktionen
+
+test('Konfetti laeuft sofort, ohne Freigabe', () => {
+  const e = entscheiden(einloesung({ rewardTitle: 'Konfetti', text: '' }), CFG);
+  assert.deepEqual(e, { art: 'ausfuehren', aktion: 'konfetti', plan: CFG.belohnungen[0] });
+});
+
+test('der Kill-Switch haelt alles an', () => {
+  const e = entscheiden(einloesung({ rewardTitle: 'Konfetti' }), { ...CFG, enabled: false });
+  assert.equal(e.art, 'abgelehnt');
+});
+
+// ---------------------------------------------------------------- TTS
+
+test('mit Freigabe geht Text in die Warteschlange, nicht ins Bild', () => {
+  const e = entscheiden(einloesung(), CFG);
+  assert.deepEqual(e, { art: 'warteschlange', text: 'moin zusammen' });
+});
+
+test('ohne Freigabe wird sofort vorgelesen', () => {
+  const e = entscheiden(einloesung(), { ...CFG, tts: { ...CFG.tts, freigabe: false } });
+  assert.deepEqual(e, { art: 'vorlesen', text: 'moin zusammen' });
+});
+
+test('die Einstellung an der Belohnung schlaegt die allgemeine', () => {
+  const cfg: RewardsConfig = {
+    ...CFG,
+    belohnungen: [{ titel: 'Vorlesen', aktion: 'tts', freigabe: false }],
+    tts: { ...CFG.tts, freigabe: true },
+  };
+  assert.equal(entscheiden(einloesung(), cfg).art, 'vorlesen');
+});
+
+test('zu langer Text wird abgelehnt, nicht abgeschnitten', () => {
+  // Abgeschnitten waere schlimmer: dann wird ein halber Satz vorgelesen, und
+  // niemand weiss, ob der Rest Absicht war.
+  const e = entscheiden(einloesung({ text: 'a'.repeat(201) }), CFG);
+  assert.equal(e.art, 'abgelehnt');
+  assert.match((e as { grund: string }).grund, /zu lang/);
+});
+
+test('leerer Text wird abgelehnt', () => {
+  assert.equal(entscheiden(einloesung({ text: '   ' }), CFG).art, 'abgelehnt');
+});
+
+test('der Wortfilter greift unabhaengig von Gross-/Kleinschreibung', () => {
+  const e = entscheiden(einloesung({ text: 'guck mal HeilHitler lol' }), CFG);
+  assert.equal(e.art, 'abgelehnt');
+  assert.equal((e as { grund: string }).grund, 'Wortfilter');
+});
+
+test('der Ablehnungsgrund verraet den gefilterten Text nicht', () => {
+  // Der Grund landet im Protokoll und moeglicherweise im Chat. Er darf nicht
+  // wiederholen, was gerade herausgefiltert wurde.
+  const e = entscheiden(einloesung({ text: 'heilhitler' }), CFG);
+  assert.equal((e as { grund: string }).grund.includes('heilhitler'), false);
+});
+
+test('wer zu oft einloest, wird gebremst', () => {
+  assert.equal(entscheiden(einloesung(), CFG, 2).art, 'warteschlange');
+  assert.equal(entscheiden(einloesung(), CFG, 3).art, 'abgelehnt');
+});
+
+// ---------------------------------------------------------------- Warteschlange
+
+test('eingereiht, entnommen, weg', () => {
+  const w = createWarteschlange();
+  w.einreihen({ id: 'a', name: 'Kevin', text: 'moin', rewardId: 'r', redemptionId: 'x', seit: '' });
+  assert.equal(w.liste().length, 1);
+  assert.equal(w.entnehmen('a')?.text, 'moin');
+  assert.equal(w.liste().length, 0);
+  assert.equal(w.entnehmen('a'), undefined, 'zweimal entnehmen geht nicht');
+});
+
+test('eine volle Warteschlange wirft das Aelteste raus', () => {
+  // Eine Schlange, die bei einem Ansturm auf 300 waechst, arbeitet niemand ab.
+  const w = createWarteschlange(3);
+  for (const id of ['a', 'b', 'c', 'd']) {
+    w.einreihen({ id, name: 'x', text: id, rewardId: 'r', redemptionId: id, seit: '' });
+  }
+  assert.deepEqual(w.liste().map((x) => x.id), ['b', 'c', 'd']);
+});
+
+test('leeren gibt alles zurueck, damit die Punkte erstattet werden koennen', () => {
+  const w = createWarteschlange();
+  w.einreihen({ id: 'a', name: 'x', text: '1', rewardId: 'r', redemptionId: 'a', seit: '' });
+  w.einreihen({ id: 'b', name: 'y', text: '2', rewardId: 'r', redemptionId: 'b', seit: '' });
+  assert.equal(w.leeren().length, 2);
+  assert.equal(w.liste().length, 0);
+});
+
+// ---------------------------------------------------------------- Drossel
+
+test('die Drossel zaehlt nur das gleitende Fenster', () => {
+  const d = createDrossel(3_600_000);
+  const t = Date.parse('2026-09-17T20:00:00Z');
+  d.merken('u1', t);
+  d.merken('u1', t + 60_000);
+  assert.equal(d.zaehlen('u1', t + 120_000), 2);
+  // Zwei Stunden spaeter ist beides aus dem Fenster gefallen.
+  assert.equal(d.zaehlen('u1', t + 2 * 3_600_000), 0);
+});
+
+test('die Drossel trennt nach Person', () => {
+  const d = createDrossel();
+  d.merken('u1');
+  assert.equal(d.zaehlen('u2'), 0);
+});
+
+// ---------------------------------------------------------------- EventSub
+
+test('das EventSub-Ereignis wird in unsere Form gebracht', () => {
+  const e = ausEventSub({
+    id: 'red-1',
+    user_id: '42',
+    user_name: 'Kevin',
+    user_input: 'lies das vor',
+    redeemed_at: '2026-09-17T20:00:00Z',
+    reward: { id: 'rw-9', title: 'Vorlesen', cost: 500 },
+  });
+  assert.equal(e?.id, 'red-1');
+  assert.equal(e?.rewardId, 'rw-9');
+  assert.equal(e?.rewardTitle, 'Vorlesen');
+  assert.equal(e?.text, 'lies das vor');
+  assert.equal(e?.kosten, 500);
+});
+
+test('eine Belohnung ohne Texteingabe hat einen leeren Text, nicht "undefined"', () => {
+  // Ohne das stuende im Bild woertlich "undefined".
+  const e = ausEventSub({ id: 'x', user_id: '1', reward: { id: 'r' } });
+  assert.equal(e?.text, '');
+  assert.equal(e?.userName, '');
+});
+
+test('unvollstaendige Ereignisse werden verworfen', () => {
+  assert.equal(ausEventSub({}), null);
+  assert.equal(ausEventSub({ id: 'x' }), null);
+  assert.equal(ausEventSub({ id: 'x', user_id: '1' }), null, 'ohne Belohnung keine Aktion');
+});

apps/hub/src/rewards.ts

@@ -0,0 +1,203 @@
+// Was bei einer Kanalpunkt-Einloesung passiert.
+//
+// Der Weg: EventSub meldet die Einloesung -> hier wird sie einer Aktion aus
+// der Konfiguration zugeordnet -> die Aktion laeuft, oder sie landet in der
+// Warteschlange und wartet auf Abdus Klick.
+//
+// Zur Zuordnung: Belohnungen werden ueber ihren Titel gefunden, wahlweise
+// ueber die ID. Der Titel ist lesbar und laesst sich in der Konfiguration
+// wiedererkennen, kann sich aber aendern; die ID ist stabil, aber
+// nichtssagend. Deshalb beides, und die ID gewinnt.
+//
+// Warum TTS eine Warteschlange hat, steht in docs/projects/chat-control.md:
+// Kanalpunkte sammeln sich beim Zugucken von allein an und sind damit kaum
+// eine Huerde. Wer beliebigen Fremdtext ungeprueft vorlesen laesst, hat eine
+// Vorlesefunktion fuer Fremde im eigenen Stream.
+
+import { logger } from '#core/logger.ts';
+
+const log = logger('rewards');
+
+/** Eine Einloesung, so weit sie hier gebraucht wird. */
+export type Einloesung = {
+  id: string;
+  rewardId: string;
+  rewardTitle: string;
+  userId: string;
+  userName: string;
+  /** Der eingegebene Text. Leer, wenn die Belohnung keinen verlangt. */
+  text: string;
+  kosten: number;
+  ts: string;
+};
+
+export type AktionsArt = 'konfetti' | 'emote-regen' | 'einblendung' | 'tts' | 'nichts';
+
+export type BelohnungsPlan = {
+  /** Titel der Belohnung auf Twitch. Gross-/Kleinschreibung egal. */
+  titel?: string;
+  /** Stabile Belohnungs-ID. Gewinnt gegen den Titel. */
+  id?: string;
+  aktion: AktionsArt;
+  /** Nur bei `tts`: ohne Freigabe wird sofort vorgelesen. */
+  freigabe?: boolean;
+  /** Nur bei `tts`: laengerer Text wird abgelehnt statt gekuerzt. */
+  maxZeichen?: number;
+};
+
+export type RewardsConfig = {
+  enabled: boolean;
+  belohnungen: BelohnungsPlan[];
+  tts: {
+    /** Vorgabe fuer `freigabe`, wenn eine Belohnung nichts anderes sagt. */
+    freigabe: boolean;
+    maxZeichen: number;
+    /** Woerter, die den Text ohne Nachfrage ablehnen. */
+    blockliste: string[];
+    /** So viele Einloesungen je Person und Stunde. */
+    proPersonProStunde: number;
+  };
+};
+
+/** Was mit einer Einloesung geschehen soll. Rein, damit pruefbar. */
+export type Entscheidung =
+  | { art: 'ausfuehren'; aktion: Exclude<AktionsArt, 'tts' | 'nichts'>; plan: BelohnungsPlan }
+  | { art: 'vorlesen'; text: string }
+  | { art: 'warteschlange'; text: string }
+  | { art: 'abgelehnt'; grund: string }
+  | { art: 'unbekannt' };
+
+/** Findet den Plan zu einer Einloesung. ID gewinnt, sonst der Titel. */
+export function planFuer(
+  e: Pick<Einloesung, 'rewardId' | 'rewardTitle'>, plaene: BelohnungsPlan[],
+): BelohnungsPlan | undefined {
+  return plaene.find((p) => p.id && p.id === e.rewardId)
+    ?? plaene.find((p) => p.titel && p.titel.toLowerCase() === e.rewardTitle.toLowerCase());
+}
+
+export function entscheiden(
+  e: Einloesung,
+  cfg: RewardsConfig,
+  /** Wie oft diese Person in der letzten Stunde schon eingeloest hat. */
+  bisherigeDieseStunde = 0,
+): Entscheidung {
+  if (!cfg.enabled) return { art: 'abgelehnt', grund: 'Belohnungs-Aktionen sind ausgeschaltet' };
+
+  const plan = planFuer(e, cfg.belohnungen);
+  // Unbekannte Belohnung heisst nicht "Fehler": Abdu darf Belohnungen anlegen,
+  // die mit dem Toolkit nichts zu tun haben. Sie werden protokolliert und
+  // sonst in Ruhe gelassen.
+  if (!plan) return { art: 'unbekannt' };
+  if (plan.aktion === 'nichts') return { art: 'unbekannt' };
+
+  if (plan.aktion !== 'tts') {
+    return { art: 'ausfuehren', aktion: plan.aktion, plan };
+  }
+
+  // ---- ab hier nur noch TTS
+  const text = e.text.trim();
+  if (!text) return { art: 'abgelehnt', grund: 'kein Text eingegeben' };
+
+  const max = plan.maxZeichen ?? cfg.tts.maxZeichen;
+  if (text.length > max) {
+    // Abschneiden waere schlimmer: dann wird ein halber Satz vorgelesen, und
+    // niemand weiss, ob der Rest Absicht war.
+    return { art: 'abgelehnt', grund: `zu lang (${text.length} von ${max} Zeichen)` };
+  }
+
+  const klein = text.toLowerCase();
+  const treffer = cfg.tts.blockliste.find((w) => w && klein.includes(w.toLowerCase()));
+  if (treffer) return { art: 'abgelehnt', grund: 'Wortfilter' };
+
+  if (bisherigeDieseStunde >= cfg.tts.proPersonProStunde) {
+    return { art: 'abgelehnt', grund: 'zu viele Einloesungen in der letzten Stunde' };
+  }
+
+  const freigabe = plan.freigabe ?? cfg.tts.freigabe;
+  return freigabe ? { art: 'warteschlange', text } : { art: 'vorlesen', text };
+}
+
+// ---------------------------------------------------------------- Warteschlange
+
+export type Wartend = {
+  id: string;
+  name: string;
+  text: string;
+  /** Fuer die Erstattung, falls abgelehnt wird. */
+  rewardId: string;
+  redemptionId: string;
+  seit: string;
+};
+
+export type Warteschlange = {
+  einreihen: (w: Wartend) => void;
+  /** Gibt den Eintrag zurueck, damit der Aufrufer ihn vorlesen lassen kann. */
+  entnehmen: (id: string) => Wartend | undefined;
+  liste: () => Wartend[];
+  leeren: () => Wartend[];
+};
+
+/**
+ * Wartende Nachrichten. Bewusst im Arbeitsspeicher: was beim Neustart
+ * verloren geht, war ohnehin nicht dringend, und eine Tabelle mit fremdem
+ * Text, die niemand aufraeumt, will hier niemand.
+ */
+export function createWarteschlange(maximal = 25): Warteschlange {
+  const wartend: Wartend[] = [];
+  return {
+    einreihen(w) {
+      wartend.push(w);
+      // Laeuft sie voll, faellt das Aelteste raus. Eine Warteschlange, die
+      // bei einem Ansturm auf 300 waechst, arbeitet Abdu nie ab.
+      while (wartend.length > maximal) {
+        const weg = wartend.shift();
+        if (weg) log.warn(`Warteschlange voll, aelteste Nachricht verworfen (${weg.name})`);
+      }
+    },
+    entnehmen(id) {
+      const i = wartend.findIndex((w) => w.id === id);
+      return i === -1 ? undefined : wartend.splice(i, 1)[0];
+    },
+    liste: () => [...wartend],
+    leeren: () => wartend.splice(0, wartend.length),
+  };
+}
+
+/**
+ * Zaehlt Einloesungen je Person in einem gleitenden Fenster.
+ * Braucht keine Datenbank - nach einem Neustart faengt das Fenster neu an,
+ * und das ist bei einer Stunde Fensterbreite egal.
+ */
+export function createDrossel(fensterMs = 3_600_000) {
+  const gesehen = new Map<string, number[]>();
+  return {
+    zaehlen(userId: string, jetzt = Date.now()): number {
+      const bisher = (gesehen.get(userId) ?? []).filter((t) => jetzt - t < fensterMs);
+      gesehen.set(userId, bisher);
+      return bisher.length;
+    },
+    merken(userId: string, jetzt = Date.now()): void {
+      const bisher = (gesehen.get(userId) ?? []).filter((t) => jetzt - t < fensterMs);
+      bisher.push(jetzt);
+      gesehen.set(userId, bisher);
+    },
+  };
+}
+
+/** Das EventSub-Ereignis in unsere Form bringen. */
+export function ausEventSub(event: Record<string, unknown>): Einloesung | null {
+  const reward = event['reward'] as { id?: string; title?: string; cost?: number } | undefined;
+  const id = event['id'];
+  const userId = event['user_id'];
+  if (typeof id !== 'string' || typeof userId !== 'string' || !reward?.id) return null;
+  return {
+    id,
+    rewardId: reward.id,
+    rewardTitle: reward.title ?? '',
+    userId,
+    userName: String(event['user_name'] ?? event['user_login'] ?? ''),
+    text: String(event['user_input'] ?? ''),
+    kosten: Number(reward.cost ?? 0),
+    ts: String(event['redeemed_at'] ?? new Date().toISOString()),
+  };
+}

packages/twitch/src/eventsub.test.ts

@@ -0,0 +1,98 @@
+// EventSub liefert Ereignisse, die sich nicht wiederholen lassen. Wer eine
+// Nachricht falsch deutet, verpasst eine Einloesung - und der Zuschauer hat
+// die Punkte bezahlt. Deshalb ist das Deuten hier von der Verbindung getrennt
+// und einzeln geprueft.
+
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { deuten, type EventSubNachricht } from './eventsub.ts';
+
+const n = (o: unknown): EventSubNachricht => o as EventSubNachricht;
+
+test('der Willkommensgruss traegt Sitzung und Zeitlimit', () => {
+  const f = deuten(n({
+    metadata: { message_id: '1', message_type: 'session_welcome' },
+    payload: { session: { id: 'sess-1', keepalive_timeout_seconds: 10, reconnect_url: null } },
+  }));
+  assert.equal(f.art, 'sitzung');
+  assert.equal((f as { sessionId: string }).sessionId, 'sess-1');
+  // 10 s Zeitlimit plus Puffer: der Wachhund soll eine tote Verbindung finden,
+  // nicht eine langsame abwuergen.
+  assert.equal((f as { keepaliveMs: number }).keepaliveMs, 15_000);
+});
+
+test('ohne Angabe gilt Twitchs dokumentierte Vorgabe', () => {
+  const f = deuten(n({
+    metadata: { message_type: 'session_welcome' },
+    payload: { session: { id: 's', keepalive_timeout_seconds: null, reconnect_url: null } },
+  }));
+  assert.equal((f as { keepaliveMs: number }).keepaliveMs, 15_000);
+});
+
+test('eine Benachrichtigung traegt Typ und Ereignis', () => {
+  const f = deuten(n({
+    metadata: {
+      message_type: 'notification',
+      subscription_type: 'channel.channel_points_custom_reward_redemption.add',
+    },
+    payload: { event: { id: 'red-1', user_name: 'Kevin' } },
+  }));
+  assert.equal(f.art, 'ereignis');
+  assert.match((f as { typ: string }).typ, /reward_redemption\.add$/);
+  assert.equal(((f as { event: Record<string, unknown> }).event)['user_name'], 'Kevin');
+});
+
+test('fehlt der Typ in den Kopfdaten, wird er aus dem Abonnement genommen', () => {
+  const f = deuten(n({
+    metadata: { message_type: 'notification' },
+    payload: { subscription: { id: 's', type: 'channel.follow', status: 'enabled' },
+               event: { user_name: 'x' } },
+  }));
+  assert.equal((f as { typ: string }).typ, 'channel.follow');
+});
+
+test('eine Benachrichtigung ohne Ereignis wird nicht durchgereicht', () => {
+  const f = deuten(n({
+    metadata: { message_type: 'notification', subscription_type: 'channel.follow' },
+    payload: {},
+  }));
+  assert.equal(f.art, 'unbekannt');
+});
+
+test('der Umzug nennt die neue Adresse', () => {
+  const f = deuten(n({
+    metadata: { message_type: 'session_reconnect' },
+    payload: { session: { id: 's', keepalive_timeout_seconds: 10,
+                          reconnect_url: 'wss://eventsub.wss.twitch.tv/ws?id=neu' } },
+  }));
+  assert.deepEqual(f, { art: 'umziehen', url: 'wss://eventsub.wss.twitch.tv/ws?id=neu' });
+});
+
+test('ein Umzug ohne Adresse wird ignoriert statt zu einer leeren Verbindung zu fuehren', () => {
+  const f = deuten(n({
+    metadata: { message_type: 'session_reconnect' },
+    payload: { session: { id: 's', keepalive_timeout_seconds: 10, reconnect_url: null } },
+  }));
+  assert.equal(f.art, 'unbekannt');
+});
+
+test('ein Entzug wird als solcher erkannt', () => {
+  // Passiert, wenn Abdu die App-Freigabe zurueckzieht. Wiederverbinden hilft
+  // dagegen nichts, deshalb darf es nicht wie ein Netzwerkfehler aussehen.
+  const f = deuten(n({
+    metadata: { message_type: 'revocation' },
+    payload: { subscription: { id: 's', type: 'channel.follow', status: 'authorization_revoked' } },
+  }));
+  assert.deepEqual(f, { art: 'entzogen', typ: 'channel.follow', grund: 'authorization_revoked' });
+});
+
+test('ein Lebenszeichen ist ein Lebenszeichen', () => {
+  assert.deepEqual(deuten(n({ metadata: { message_type: 'session_keepalive' }, payload: {} })),
+    { art: 'lebenszeichen' });
+});
+
+test('Unsinn wird verworfen, nicht geraten', () => {
+  for (const muell of [{}, { metadata: {} }, { metadata: { message_type: 'gibtsnicht' }, payload: {} }]) {
+    assert.equal(deuten(n(muell)).art, 'unbekannt');
+  }
+});

packages/twitch/src/eventsub.ts

@@ -0,0 +1,229 @@
+// EventSub ueber WebSocket. Beschlossen in ADR-0003, gebaut erst jetzt.
+//
+// Warum ueberhaupt, wo der Chat doch schon da ist: Kanalpunkt-Einloesungen
+// kommen ueber IRC nur dann an, wenn die Belohnung eine Texteingabe verlangt -
+// und selbst dann ohne Einloese-ID. Ohne die ID lassen sich Punkte nicht
+// zurueckgeben, wenn eine Aktion schiefgeht. Wer Punkte nimmt und nichts
+// liefert, verliert das Vertrauen schneller als durch jeden Fehler im Bild.
+//
+// Ablauf des Protokolls:
+//
+//   verbinden -> session_welcome (traegt session.id und das Zeitlimit)
+//             -> Abonnements ueber Helix anlegen, mit dieser session.id
+//             -> notification ... notification ... session_keepalive ...
+//
+// Zwei Eigenheiten, die man kennen muss:
+//
+//   * Ein Abonnement muss innerhalb von 10 Sekunden nach dem Willkommensgruss
+//     stehen, sonst schliesst Twitch die Verbindung wieder.
+//   * `session_reconnect` nennt eine neue Adresse. Die alte Verbindung bleibt
+//     offen, bis die neue steht - erst verbinden, dann schliessen, sonst
+//     entsteht genau in dieser Luecke ein verpasstes Ereignis.
+
+import { logger } from '#core/logger.ts';
+import { forever } from '#core/retry.ts';
+
+const log = logger('eventsub');
+
+export const EVENTSUB_URL = 'wss://eventsub.wss.twitch.tv/ws';
+
+/**
+ * Wie lange ohne ein Lebenszeichen gewartet wird, bevor neu verbunden wird.
+ * Twitch nennt sein eigenes Zeitlimit im Willkommensgruss (ueblich 10 s); wir
+ * legen Luft drauf, damit ein kurzer Netzhaenger nicht sofort neu verbindet.
+ */
+const KEEPALIVE_PUFFER_MS = 5_000;
+
+export type EventSubNachricht = {
+  metadata: {
+    message_id: string;
+    message_type: 'session_welcome' | 'session_keepalive' | 'notification'
+      | 'session_reconnect' | 'revocation';
+    subscription_type?: string;
+    subscription_version?: string;
+  };
+  payload: {
+    session?: { id: string; keepalive_timeout_seconds: number | null; reconnect_url: string | null };
+    subscription?: { id: string; type: string; status: string };
+    event?: Record<string, unknown>;
+  };
+};
+
+/** Ein Abonnement, so wie Helix es entgegennimmt. */
+export type Abonnement = {
+  type: string;
+  version: string;
+  condition: Record<string, string>;
+};
+
+export type EventSubHandlers = {
+  /** Die Sitzung steht. Hier gehoeren die Abonnements angelegt. */
+  onSession: (sessionId: string) => Promise<void>;
+  onNotification: (typ: string, event: Record<string, unknown>) => void;
+  /**
+   * Twitch hat ein Abonnement entzogen - etwa weil das Token zurueckgezogen
+   * wurde. Das ist kein Netzwerkproblem und geht durch Wiederverbinden nicht
+   * weg, deshalb bekommt es einen eigenen Weg.
+   */
+  onRevocation?: (typ: string, grund: string) => void;
+  onConnected?: () => void;
+  onClosed?: () => void;
+};
+
+/**
+ * Was aus einer eingehenden Nachricht folgen soll. Ausgelagert und rein,
+ * damit sich die Faelle pruefen lassen, die im Betrieb selten sind und dann
+ * mitten im Stream zuschlagen.
+ */
+export type Folge =
+  | { art: 'sitzung'; sessionId: string; keepaliveMs: number }
+  | { art: 'ereignis'; typ: string; event: Record<string, unknown> }
+  | { art: 'umziehen'; url: string }
+  | { art: 'entzogen'; typ: string; grund: string }
+  | { art: 'lebenszeichen' }
+  | { art: 'unbekannt' };
+
+export function deuten(n: EventSubNachricht): Folge {
+  switch (n.metadata?.message_type) {
+    case 'session_welcome': {
+      const s = n.payload?.session;
+      if (!s?.id) return { art: 'unbekannt' };
+      // Ohne Angabe 10 Sekunden annehmen, so wie Twitch es dokumentiert.
+      const sek = s.keepalive_timeout_seconds ?? 10;
+      return { art: 'sitzung', sessionId: s.id, keepaliveMs: sek * 1000 + KEEPALIVE_PUFFER_MS };
+    }
+    case 'session_reconnect': {
+      const url = n.payload?.session?.reconnect_url;
+      return url ? { art: 'umziehen', url } : { art: 'unbekannt' };
+    }
+    case 'notification': {
+      const typ = n.metadata.subscription_type ?? n.payload?.subscription?.type ?? '';
+      const event = n.payload?.event;
+      if (!typ || !event) return { art: 'unbekannt' };
+      return { art: 'ereignis', typ, event };
+    }
+    case 'revocation':
+      return {
+        art: 'entzogen',
+        typ: n.payload?.subscription?.type ?? 'unbekannt',
+        grund: n.payload?.subscription?.status ?? 'unbekannt',
+      };
+    case 'session_keepalive':
+      return { art: 'lebenszeichen' };
+    default:
+      return { art: 'unbekannt' };
+  }
+}
+
+export type EventSubClient = {
+  run: (signal: AbortSignal) => Promise<void>;
+  isConnected: () => boolean;
+};
+
+export function createEventSub(handlers: EventSubHandlers): EventSubClient {
+  let verbunden = false;
+
+  return {
+    run: (signal) => forever(
+      async (sig) => { await einmalVerbinden(EVENTSUB_URL, handlers, sig, (v) => { verbunden = v; }); },
+      {
+        signal,
+        onWait: (ms, n) =>
+          log.warn(`EventSub getrennt, neuer Versuch in ${Math.round(ms / 1000)} s (${n}.)`),
+      },
+    ),
+    isConnected: () => verbunden,
+  };
+}
+
+function einmalVerbinden(
+  url: string,
+  handlers: EventSubHandlers,
+  signal: AbortSignal,
+  setzeStatus: (v: boolean) => void,
+): Promise<void> {
+  return new Promise<void>((_erfuellt, ablehnen) => {
+    const ws = new WebSocket(url);
+    let wachhund: ReturnType<typeof setTimeout> | null = null;
+    let umzug: WebSocket | null = null;
+    // Twitch nennt sein Zeitlimit im Willkommensgruss. Bis dahin grosszuegig.
+    let frist = 30_000;
+
+    const abbrechen = (warum: string): void => {
+      if (wachhund) clearTimeout(wachhund);
+      setzeStatus(false);
+      handlers.onClosed?.();
+      try { ws.close(); } catch { /* schon zu */ }
+      try { umzug?.close(); } catch { /* schon zu */ }
+      ablehnen(new Error(warum));
+    };
+
+    signal.addEventListener('abort', () => abbrechen('abgebrochen'), { once: true });
+
+    /** Kommt zu lange nichts, ist die Verbindung tot, ohne es zu melden. */
+    const wachhundStellen = (ms: number): void => {
+      if (wachhund) clearTimeout(wachhund);
+      wachhund = setTimeout(() => abbrechen('kein Lebenszeichen mehr'), ms);
+    };
+
+    ws.addEventListener('open', () => wachhundStellen(30_000));
+
+    ws.addEventListener('message', (ev) => {
+      let nachricht: EventSubNachricht;
+      try { nachricht = JSON.parse(String(ev.data)); } catch { return; }
+      const folge = deuten(nachricht);
+
+      switch (folge.art) {
+        case 'sitzung':
+          frist = folge.keepaliveMs;
+          wachhundStellen(frist);
+          setzeStatus(true);
+          handlers.onConnected?.();
+          // Die Abonnements muessen jetzt angelegt werden. Scheitert das,
+          // haelt Twitch die Verbindung ohnehin nicht lange.
+          void handlers.onSession(folge.sessionId).catch((err: Error) => {
+            log.error(`Abonnements nicht angelegt: ${err.message}`);
+            abbrechen('Abonnements fehlgeschlagen');
+          });
+          break;
+
+        case 'lebenszeichen':
+          wachhundStellen(frist);
+          break;
+
+        case 'ereignis':
+          wachhundStellen(frist);
+          try { handlers.onNotification(folge.typ, folge.event); }
+          catch (err) { log.error(`Ereignis nicht verarbeitet: ${(err as Error).message}`); }
+          break;
+
+        case 'umziehen': {
+          // Erst die neue Verbindung aufbauen, dann die alte schliessen.
+          // Andersherum faellt genau dazwischen ein Ereignis unter den Tisch.
+          log.info('Twitch bittet um Umzug auf eine neue Verbindung');
+          umzug = new WebSocket(folge.url);
+          umzug.addEventListener('open', () => {
+            log.info('Umzug steht, alte Verbindung wird geschlossen');
+            abbrechen('umgezogen');
+          });
+          umzug.addEventListener('error', () => {
+            log.warn('Umzug fehlgeschlagen, bleibe auf der alten Verbindung');
+            umzug = null;
+          });
+          break;
+        }
+
+        case 'entzogen':
+          log.error(`Abonnement entzogen: ${folge.typ} (${folge.grund})`);
+          handlers.onRevocation?.(folge.typ, folge.grund);
+          break;
+
+        default:
+          break;
+      }
+    });
+
+    ws.addEventListener('error', () => abbrechen('WebSocket-Fehler'));
+    ws.addEventListener('close', (ev) => abbrechen(`Verbindung geschlossen (${ev.code})`));
+  });
+}