diff --git a/apps/hub/src/rewards.test.ts b/apps/hub/src/rewards.test.ts
new file mode 100644
index 0000000..4a4442d
--- /dev/null
+++ b/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');
+});
diff --git a/apps/hub/src/rewards.ts b/apps/hub/src/rewards.ts
new file mode 100644
index 0000000..e11b54b
--- /dev/null
+++ b/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()),
+  };
+}
diff --git a/packages/twitch/src/eventsub.test.ts b/packages/twitch/src/eventsub.test.ts
new file mode 100644
index 0000000..2349f1b
--- /dev/null
+++ b/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');
+  }
+});
diff --git a/packages/twitch/src/eventsub.ts b/packages/twitch/src/eventsub.ts
new file mode 100644
index 0000000..299416c
--- /dev/null
+++ b/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})`));
+  });
+}