abdulive/apps/ingest/src/index.ts

// Ingest-Daemon. Liest Chat und Zuschauerzahlen, schreibt beides in die
// Datenbank und haelt die Einrichtungsseite bereit.
//
// Stand S2: IRC, Helix-Polling, obs-websocket und der !clip-Ausloeser.
// EventSub kommt spaeter - fuer den Clip-Weg reicht IRC plus Polling.
// Siehe docs/projects/00-ingest.md.

import { loadConfig, loadLayout, paths } 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 { createObs } from '#obs/client.ts';
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 { createClipper } from './clips.ts';
import { createRenderer } from './render.ts';
import { createSpikeDetector } from './spike.ts';
import { startServer } from './server.ts';

const log = logger('ingest');

export async function runIngest(): Promise<void> {
  const cfg = loadConfig();
  const db = openDb();
  const repo = repos(db);
  const spikes = createSpikeDetector(cfg.ingest.spike);

  const state = {
    startedAt: new Date().toISOString(),
    streamId: null as string | null,
    broadcasterId: null as string | null,
    lastSeenLive: 0,
    viewers: 0,
    chatConnected: false,
    chatAnonymous: true,
    obsConnected: false,
    obsScene: null as string | null,
    messages: 0,
    spikeCandidates: 0,
    clipsTriggered: 0,
    lastError: null as string | null,
  };

  const abort = new AbortController();
  const stop = (signal: string) => {
    log.info(`${signal} empfangen, fahre herunter`);
    abort.abort();
    setTimeout(() => { closeDb(); process.exit(0); }, 500);
  };
  process.on('SIGINT', () => stop('SIGINT'));
  process.on('SIGTERM', () => stop('SIGTERM'));

  // ---- OBS (Wiedergabepuffer, Szenenwechsel)
  const obs = createObs(
    cfg.obs.url,
    () => readSecrets().obs?.password,
    {
      onIdentified: () => { state.obsConnected = true; },
      onClosed: () => { state.obsConnected = false; },
      onSceneChanged: (scene) => {
        state.obsScene = scene;
        repo.event('obs.scene_changed', { scene }, state.streamId);
      },
      onReplaySaved: (hostPath) => { void clipper.onReplaySaved(hostPath); },
    }
  );

  const layout = loadLayout();
  const profile = layout.profiles['default'];
  if (!profile) throw new Error('config/layout.yaml hat kein Profil "default"');

  const renderer = createRenderer({
    cfg: {
      profile,
      tailSeconds: cfg.clipper.tailSeconds,
      outputTrack: cfg.clipper.audio.outputTrack,
      encoder: cfg.clipper.ffmpeg.encoder,
      preset: cfg.clipper.ffmpeg.preset,
      crf: cfg.clipper.ffmpeg.crf,
      niceness: cfg.clipper.ffmpeg.niceness,
      renderWhileLive: cfg.clipper.renderWhileLive,
    },
    repo,
    workDir: paths.work,
    isLive: () => state.streamId !== null,
  });

  const clipper = createClipper({
    cfg: {
      enabled: cfg.clipper.enabled,
      alsoCreateTwitchClip: cfg.clipper.alsoCreateTwitchClip,
      cooldownGlobalSeconds: cfg.clipper.cooldownGlobalSeconds,
      cooldownUserSeconds: cfg.clipper.cooldownUserSeconds,
      followersOnly: cfg.clipper.followersOnly,
      ...(process.env.OBS_REPLAY_DIR ? { replayDirHost: process.env.OBS_REPLAY_DIR } : {}),
      // Nativ sieht die Anwendung denselben Pfad wie OBS. Leer heisst:
      // nicht uebersetzen (siehe #obs/replay-path.ts).
      replayDirContainer: isWindows ? '' : paths.replay,
      clipsDir: paths.clips,
      workDir: paths.work,
    },
    repo,
    obs,
    say: (text) => irc.say(text),
    currentStreamId: () => state.streamId,
    broadcasterId: () => state.broadcasterId,
    onRawReady: (job) => renderer.enqueue(job),
  });

  startServer({
    channel: cfg.channel.login,
    port: cfg.dashboard.port,
    health: () => ({
      ...state,
      db: repo.stats(),
      spikeArmed: spikes.isArmed(),
      clipsWaitingForFile: clipper.pendingCount(),
      renderQueue: renderer.pending(),
      rendering: renderer.currentJob(),
      note: state.chatAnonymous
        ? 'Chat wird anonym mitgelesen. Zum Senden und fuer Helix: Einrichtung unter /setup.'
        : undefined,
    }),
  });

  // ---- Chat
  const irc = createIrc(cfg.channel.login, {
    onConnected: (anonymous) => {
      state.chatConnected = true;
      state.chatAnonymous = anonymous;
    },
    onMessage: (msg) => {
      try {
        const { streamsChatted } = 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);

        if (msg.isFirst) {
          repo.event('chat.first_message', { login: msg.login }, state.streamId, msg.userId, msg.ts);
          log.info(`Erstchatter: ${msg.displayName}`);
        } else if (streamsChatted === 1) {
          log.debug(`Erster Stream fuer ${msg.displayName}`);
        }

        if (msg.isCommand && /^!clip\b/i.test(msg.text)) {
          state.clipsTriggered++;
          void clipper.handleCommand(msg.userId, msg.displayName, msg.isMod || msg.isBroadcaster)
            .then((reply) => { if (reply) irc.say(`@${msg.displayName} ${reply}`); });
        }

        const spike = spikes.push(msg);
        if (spike) {
          state.spikeCandidates++;
          repo.event('chat.spike', spike, state.streamId, null, msg.ts);
          log.info(`Spitze im Chat: Punktzahl ${spike.score} ` +
            `(${spike.rate} Nachrichten von ${spike.chatters} Leuten, Basislinie ${spike.baseline})` +
            (spike.overThreshold ? '' : ' - nur protokolliert, Ausloesung ist aus'));
        }
      } catch (err) {
        state.lastError = (err as Error).message;
        log.error(`Nachricht nicht verarbeitet: ${(err as Error).message}`);
      }
    },
  });
  void irc.run(abort.signal);
  void obs.run(abort.signal);

  // ---- Zuschauerzahlen und Stream-Sessions
  void pollStream(cfg, repo, state, abort.signal);

  log.info(`Ingest laeuft fuer #${cfg.channel.login}`);
  await new Promise<void>((resolve) => abort.signal.addEventListener('abort', () => resolve()));
}

type State = {
  streamId: string | null; broadcasterId: string | null;
  lastSeenLive: number; viewers: number; lastError: string | null;
};

async function pollStream(
  cfg: ReturnType<typeof loadConfig>,
  repo: ReturnType<typeof repos>,
  state: State,
  signal: AbortSignal
): Promise<void> {
  const gapMs = cfg.ingest.sessionGapMinutes * 60_000;
  let warnedMissingAuth = false;

  // Ein laufender Stream wird beim Start uebernommen, damit ein Neustart
  // mitten im Stream keine zweite Sitzung anlegt.
  const open = repo.latestOpenStream();
  if (open) {
    state.streamId = open.id;
    state.lastSeenLive = Date.now();
    log.info(`Laufende Sitzung uebernommen: ${open.id} seit ${open.started_at}`);
  }

  while (!signal.aborted) {
    try {
      if (!readSecrets().tokens.broadcaster) {
        if (!warnedMissingAuth) {
          log.warn('Kein Broadcaster-Token - Zuschauerzahlen und Stream-Sitzungen bleiben aus. ' +
            'Einrichtung unter /setup.');
          warnedMissingAuth = true;
        }
      } else {
        warnedMissingAuth = false;
        const userId = state.broadcasterId || cfg.channel.userId ||
          (await helix.userByLogin(cfg.channel.login))?.id || '';
        // Einmal aufgeloest, danach gemerkt - spart einen Aufruf je Minute
        if (userId) state.broadcasterId = userId;
        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);
              repo.event('stream.online',
                { title: live.title, category: live.game_name }, live.id);
              log.info(`Stream online: "${live.title}" (${live.game_name})`);
            }
          }
          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) {
      state.lastError = (err as Error).message;
      log.error(`Abfrage fehlgeschlagen: ${(err as Error).message}`);
    }
    try { await sleep(cfg.ingest.viewerPollSeconds * 1000, signal); } catch { return; }
  }
}