abdulive/apps/hub/src/index.ts

// 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 { createBadges } from '#twitch/badges.ts';
import { createIrc, type UserNotice } from '#twitch/irc.ts';
import { streamWechsel } from '#twitch/stream-state.ts';
import { piperFehlt, sprechen, stimmenPfad } from '#media/piper.ts';
import {
  ausEventSub, createDrossel, createWarteschlange, entscheiden,
} from './rewards.ts';
import { createAudioAblage } from './audio-ablage.ts';
import { createLiegestuetze } from './liegestuetze.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();
  const liegestuetze = createLiegestuetze(db);
  // Gleich zu Beginn in den Bus, sonst zeigt eine frisch verbundene
  // Browser-Source 0, bis die naechste Einloesung kommt.
  bus.publish({ kind: 'liegestuetze', ...liegestuetze.stand() });

  /**
   * Eine Einloesung abschliessen (FULFILLED) oder die Punkte zurueckgeben
   * (CANCELED). Geht nur bei Belohnungen, die das Toolkit selbst angelegt hat
   * (`abdulive rewards:anlegen`) - sonst lehnt Twitch ab, und es bleibt bei
   * einer Warnung.
   */
  const einloesungSetzen = async (
    rewardId: string, redemptionId: string, status: 'FULFILLED' | 'CANCELED',
  ): Promise<void> => {
    const was = status === 'CANCELED' ? 'Punkte nicht erstattet' : 'Einloesung nicht abgeschlossen';
    if (!state.broadcasterId) return;
    if (!helix.hasScope('channel:manage:redemptions')) {
      log.warn(`${was}: dem verbundenen Konto fehlt ` +
        'channel:manage:redemptions. Einmal neu verbinden unter /setup.');
      return;
    }
    try {
      await helix.setRedemptionStatus(state.broadcasterId, rewardId, redemptionId, status);
    } catch (err) {
      log.warn(`${was}: ${(err as Error).message}`);
    }
  };
  const erstatten = (rewardId: string, redemptionId: string): Promise<void> =>
    einloesungSetzen(rewardId, redemptionId, 'CANCELED');

  // ---- Vorlesen. Die Stimme entsteht hier im Hub (Piper), nicht im Browser:
  // dort haengt sie am Rechner, der streamt, und kam in OBS amerikanisch.
  const stimme = {
    model: stimmenPfad(cfg.rewards.tts.stimme),
    ...(typeof cfg.rewards.tts.sprecher === 'number' ? { speaker: cfg.rewards.tts.sprecher } : {}),
  };
  const audio = createAudioAblage();
  const stimmeFehlt = piperFehlt(stimme);
  if (stimmeFehlt.length) {
    log.warn(`Vorlesen ohne Stimme - es fehlt: ${stimmeFehlt.join(', ')}. ` +
      'Einrichten mit scripts/install-piper.sh. Bis dahin wird nur angezeigt.');
  }

  const vorlesen = async (
    id: string, name: string, text: string,
    /** Fehlt beim Testknopf - dann gibt es bei Twitch nichts abzuschliessen. */
    einloesung?: { rewardId: string; redemptionId: string },
  ): Promise<void> => {
    let mitTon = false;
    try {
      audio.ablegen(id, await sprechen(text, stimme));
      mitTon = true;
    } catch (err) {
      // Lieber stumm anzeigen als gar nicht: wer Punkte ausgegeben hat, soll
      // seine Nachricht wenigstens im Bild sehen.
      log.warn(`Nicht gesprochen, nur angezeigt: ${(err as Error).message}`);
    }
    bus.publish({ kind: 'tts', id, name, text, audio: mitTon });
    log.info(`Wird vorgelesen (${name}): ${text.slice(0, 60)}`);
    // Auch ohne Ton abgeschlossen: angezeigt wurde es, und eine Einloesung,
    // die ewig in Twitchs Warteschlange haengt, raeumt niemand auf.
    if (einloesung) void einloesungSetzen(einloesung.rewardId, einloesung.redemptionId, 'FULFILLED');
  };

  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 });
        if (ent.aktion === 'liegestuetze') {
          const dazu = ent.plan.anzahl ?? 5;
          bus.publish({ kind: 'liegestuetze', ...liegestuetze.aendern(dazu), dazu, name: e.userName });
        }
        log.info(`${e.userName} hat "${e.rewardTitle}" eingeloest -> ${ent.aktion}`);
        void einloesungSetzen(e.rewardId, e.id, 'FULFILLED');
        break;

      case 'vorlesen':
        void vorlesen(ulid(), e.userName, ent.text, { rewardId: e.rewardId, redemptionId: e.id });
        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);

  // ---- Abzeichen-Bilder fuer den Chat. Twitch aendert sie selten, ein
  // neues Abo-Abzeichen des Kanals ist nach spaetestens sechs Stunden da.
  const badges = createBadges();
  void (async () => {
    while (!abort.signal.aborted) {
      if (helix.canReadStream()) {
        const id = state.broadcasterId || cfg.channel.userId ||
          (await helix.userByLogin(cfg.channel.login).catch(() => null))?.id || null;
        if (id) state.broadcasterId = id;
        await badges.laden(id);
      }
      try { await sleep(6 * 60 * 60_000, abort.signal); } catch { return; }
    }
  })();

  // ---- 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,
          badges: badges.aufloesen(msg.badges),
          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,
        liegestuetze: liegestuetze.stand(),
        ttsStimme: stimmeFehlt.length ? `fehlt: ${stimmeFehlt.join(', ')}` : 'bereit',
        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;
        void vorlesen(w.id, w.name, w.text, { rewardId: w.rewardId, redemptionId: w.redemptionId });
        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;
      },
      probe: (name, text) => { void vorlesen(ulid(), name, text); },
      audio: (id) => audio.holen(id),
      still: () => {
        bus.publish({ kind: 'still' });
        audio.leeren();
        const alle = warteschlange.leeren();
        for (const w of alle) void erstatten(w.rewardId, w.redemptionId);
        log.warn('Panikschalter: Vorlesen gestoppt und Warteschlange geleert');
      },
    },
    liegestuetze: {
      stand: () => liegestuetze.stand(),
      aendern: (n) => {
        const neu = liegestuetze.aendern(n);
        bus.publish({ kind: 'liegestuetze', ...neu });
        log.info(`Liegestuetze von Hand ${n > 0 ? '+' : ''}${n} -> ` +
          `${neu.offen} offen, ${neu.gemacht} gemacht`);
        return neu;
      },
      setzen: (n) => {
        const neu = liegestuetze.setzen(n);
        bus.publish({ kind: 'liegestuetze', ...neu });
        log.info(`Liegestuetze von Hand gesetzt -> ${neu.offen} offen`);
        return neu;
      },
      gemachtSetzen: (n) => {
        const neu = liegestuetze.gemachtSetzen(n);
        bus.publish({ kind: 'liegestuetze', ...neu });
        log.info(`Liegestuetze gemacht von Hand gesetzt -> ${neu.gemacht}`);
        return neu;
      },
    },
  });
  void server;

  // ---- Zuschauerzahlen und Stream-Sitzungen
  void pollStream(cfg, repo, state, bus, abort.signal, (stream) => {
    announcer.announce(stream);
    // Nur bei einem wirklich neuen Stream, nicht beim Neustart des Hubs
    // mitten im Stream: dann wird die Sitzung uebernommen, nicht gestartet.
    bus.publish({ kind: 'liegestuetze', ...liegestuetze.neuerStream() });
  });

  // ---- 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; }
  }
}