diff --git a/apps/ingest/src/index.ts b/apps/ingest/src/index.ts index 4162570..949c2a0 100644 --- a/apps/ingest/src/index.ts +++ b/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 { 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 { 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 { ...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 { 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((resolve) => abort.signal.addEventListener('abort', () => resolve())); @@ -185,14 +206,16 @@ export async function runIngest(): Promise { 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, repo: ReturnType, state: State, - signal: AbortSignal + signal: AbortSignal, + /** Genau einmal je neu erkanntem Stream. Fortsetzungen zaehlen nicht. */ + onOnline: (stream: StreamInfo) => void ): Promise { 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(); + // 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 }, 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) { - 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) { diff --git a/packages/twitch/src/helix.ts b/packages/twitch/src/helix.ts index 0eee57c..bff5702 100644 --- a/packages/twitch/src/helix.ts +++ b/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(role: Role, endpoint: string, init: RequestInit = {}): Promise { - 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( + role: Role, endpoint: string, init: RequestInit = {}, opts: CallOpts = {} +): Promise { + 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(role: Role, endpoint: string, init: RequestInit = {}): Pr export const helix = { async userByLogin(login: string): Promise { - 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 { - 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 { 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 }, + 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 { + 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' }); diff --git a/packages/twitch/src/stream-state.test.ts b/packages/twitch/src/stream-state.test.ts new file mode 100644 index 0000000..e58b21e --- /dev/null +++ b/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 { + 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 = [ + ...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'); +}); diff --git a/packages/twitch/src/stream-state.ts b/packages/twitch/src/stream-state.ts new file mode 100644 index 0000000..f0ed8e0 --- /dev/null +++ b/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' }; +}