S1: Einrichtungs-Assistent, Twitch-Auth, Datenbank und Chat-Aufnahme
Zugangsdaten kommen jetzt aus der App statt aus einer .env: ein Einrichtungs-Assistent auf http://localhost:4792/setup nimmt Client-ID, Secret und OBS-Passwort entgegen, der OAuth-Rueckruf laeuft auf denselben Server. Gespeichert wird verschlueselt (AES-256-GCM) in einem Volume, bewusst nicht in der Datenbank - die wird auf die Windows-Seite gesichert. Siehe ADR-0005. Technikbasis vereinfacht: Node 24 bringt node:sqlite, WebSocket, fetch und die Ausfuehrung von TypeScript mit. Das Projekt hat damit null Laufzeitabhaengigkeiten und keinen Build-Schritt; pnpm-Workspaces und die Paketmanifeste sind entfallen, Importe laufen ueber die imports-Map der Wurzel-package.json. erasableSyntaxOnly verhindert Syntax, die das Type Stripping stillschweigend entfernen wuerde. Siehe Nachtrag 2 in ADR-0001. Enthalten: - packages/core: Config, Layout, Logger mit Secret-Maskierung, Retry, YAML - packages/db: Migrationen, Repositories, alle SQL-Anweisungen an einem Ort - packages/twitch: Secret-Speicher, OAuth, Helix mit Token-Bucket, IRC - apps/ingest: Daemon, Einrichtungsseite, Spike-Erkennung - 20 Tests (IRC-Parser, Spike-Detektor, Datenbankschicht) Der Chat verbindet sich anonym, solange nichts hinterlegt ist - damit laesst sich die Datenaufnahme pruefen, bevor die Twitch-Apps existieren. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013VTGjd92LsDDWF3frd9FTB
This commit is contained in:
ZeasyAlexandClaude Opus 5 committed
1 parent
4ceb61f095
commit
d07c18441b
82 files changed
+2775
-838
No files matched your search
+176
-2
@@ -1,2 +1,176 @@
|
||||
// @abdulive/ingest — siehe docs/projects/00-ingest.md
|
||||
export {};
|
||||
// Ingest-Daemon. Liest Chat und Zuschauerzahlen, schreibt beides in die
|
||||
// Datenbank und haelt die Einrichtungsseite bereit.
|
||||
//
|
||||
// Stand S1: IRC + Helix-Polling. EventSub und OBS kommen in S2.
|
||||
// Siehe docs/projects/00-ingest.md.
|
||||
|
||||
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 { helix } from '#twitch/helix.ts';
|
||||
import { createIrc } from '#twitch/irc.ts';
|
||||
import { readSecrets } from '#twitch/secrets.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,
|
||||
lastSeenLive: 0,
|
||||
viewers: 0,
|
||||
chatConnected: false,
|
||||
chatAnonymous: true,
|
||||
messages: 0,
|
||||
spikeCandidates: 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'));
|
||||
|
||||
startServer({
|
||||
channel: cfg.channel.login,
|
||||
port: cfg.dashboard.port,
|
||||
health: () => ({
|
||||
...state,
|
||||
db: repo.stats(),
|
||||
spikeArmed: spikes.isArmed(),
|
||||
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}`);
|
||||
}
|
||||
|
||||
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);
|
||||
|
||||
// ---- 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; 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 = cfg.channel.userId ||
|
||||
(await helix.userByLogin(cfg.channel.login))?.id || '';
|
||||
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; }
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user