feat: add native Windows broadcaster and live score modules
- streamer/windows: vash-broadcast.exe (setup/login/start/stop/status/doctor) with per-instance OBS, tosu and FFmpeg, a supervised worker, OCR room selection and OBS scene reconciliation; built and tested on Linux only - streamer/relay.ts shared with Vash TV; Cloudflare Access proxy for dev/staging - worker status validation and live score helpers for the site - STREAMING_HANDOFF.md and streamer/ALPHA_TEST.md record what is unverified Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
550
streamer/relay.ts
Normal file
550
streamer/relay.ts
Normal file
@@ -0,0 +1,550 @@
|
||||
/**
|
||||
* Going on and off air for one channel: a source produces one MPEG-TS
|
||||
* stream; it is copied to one small ffmpeg per destination (Twitch,
|
||||
* YouTube, …) so a failing destination reconnects without touching the
|
||||
* others, and cut into 2-second HLS segments that are uploaded to storage:
|
||||
* the live stream on the Vash site and the recording afterwards.
|
||||
*
|
||||
* Vash TV uses it with its own screen encoder; the Windows worker with an
|
||||
* RTMP ingest that OBS streams into.
|
||||
*/
|
||||
import { mkdir, readdir, readFile, rm } from 'node:fs/promises';
|
||||
import { join } from 'node:path';
|
||||
import type { Subprocess } from 'bun';
|
||||
import type { StreamConfig, VashApi } from './vash-api';
|
||||
|
||||
export type Log = (...parts: unknown[]) => void;
|
||||
|
||||
export interface OutputStatus {
|
||||
id: string;
|
||||
label: string;
|
||||
state: Output['state'];
|
||||
error: string | null;
|
||||
}
|
||||
|
||||
/** One consumer of the encoded stream: a destination, or the recorder. */
|
||||
export class Output {
|
||||
state: 'starting' | 'live' | 'reconnecting' | 'error' = 'starting';
|
||||
error: string | null = null;
|
||||
private process: Subprocess<'pipe', 'ignore', 'pipe'> | null = null;
|
||||
private since = 0;
|
||||
private queue: Uint8Array[] = [];
|
||||
private queued = 0;
|
||||
private writing = false;
|
||||
private attempts = 0;
|
||||
private stopped = false;
|
||||
|
||||
constructor(
|
||||
readonly id: string,
|
||||
readonly label: string,
|
||||
private args: string[],
|
||||
private ffmpeg: string,
|
||||
private log: Log,
|
||||
/** The destination URL, to notice when its settings change. */
|
||||
readonly url: string | null = null
|
||||
) {
|
||||
this.spawn();
|
||||
}
|
||||
|
||||
private spawn() {
|
||||
if (this.stopped) return;
|
||||
const process = Bun.spawn(
|
||||
[
|
||||
this.ffmpeg,
|
||||
'-hide_banner',
|
||||
'-loglevel',
|
||||
'error',
|
||||
'-f',
|
||||
'mpegts',
|
||||
'-i',
|
||||
'pipe:0',
|
||||
...this.args
|
||||
],
|
||||
{ stdin: 'pipe', stdout: 'ignore', stderr: 'pipe', windowsHide: true }
|
||||
);
|
||||
this.process = process;
|
||||
this.since = Date.now();
|
||||
this.queue = [];
|
||||
this.queued = 0;
|
||||
void (async () => {
|
||||
let last = '';
|
||||
for await (const chunk of process.stderr)
|
||||
last = new TextDecoder().decode(chunk).trim().split('\n').pop() || last;
|
||||
const code = await process.exited;
|
||||
if (this.stopped || this.process !== process) return;
|
||||
this.error = last || `ffmpeg exited with ${code}`;
|
||||
this.state = this.attempts > 3 ? 'error' : 'reconnecting';
|
||||
const wait = Math.min(30_000, 2000 * 2 ** this.attempts++);
|
||||
this.log(`${this.label}: ${this.error}; retrying in ${Math.round(wait / 1000)}s`);
|
||||
setTimeout(() => this.spawn(), wait);
|
||||
})();
|
||||
}
|
||||
|
||||
/**
|
||||
* Queues bytes in order. Dropping any would corrupt the MPEG-TS timestamps,
|
||||
* so a destination that falls far behind is restarted instead.
|
||||
*/
|
||||
write(chunk: Uint8Array) {
|
||||
const process = this.process;
|
||||
if (!process || this.stopped) return;
|
||||
if (this.state !== 'live' && Date.now() - this.since > 5000 && process.exitCode == null) {
|
||||
this.state = 'live';
|
||||
this.error = null;
|
||||
this.attempts = 0;
|
||||
}
|
||||
this.queue.push(chunk);
|
||||
this.queued += chunk.byteLength;
|
||||
if (this.queued > 32 * 1024 * 1024) {
|
||||
this.error = 'Destination too slow; restarting';
|
||||
this.queue = [];
|
||||
this.queued = 0;
|
||||
process.kill();
|
||||
return;
|
||||
}
|
||||
void this.drain(process);
|
||||
}
|
||||
|
||||
private async drain(process: Subprocess<'pipe', 'ignore', 'pipe'>) {
|
||||
if (this.writing) return;
|
||||
this.writing = true;
|
||||
try {
|
||||
while (this.queue.length && this.process === process) {
|
||||
const chunk = this.queue.shift()!;
|
||||
this.queued -= chunk.byteLength;
|
||||
process.stdin.write(chunk);
|
||||
const flushed = process.stdin.flush();
|
||||
if (flushed instanceof Promise) await flushed;
|
||||
}
|
||||
} catch {
|
||||
// The process is exiting; its watcher restarts it.
|
||||
} finally {
|
||||
this.writing = false;
|
||||
}
|
||||
}
|
||||
|
||||
status(): OutputStatus {
|
||||
return { id: this.id, label: this.label, state: this.state, error: this.error };
|
||||
}
|
||||
|
||||
stop() {
|
||||
this.stopped = true;
|
||||
try {
|
||||
this.process?.stdin.end();
|
||||
} catch {
|
||||
/* already closed */
|
||||
}
|
||||
const process = this.process;
|
||||
setTimeout(() => process?.kill(), 3000);
|
||||
}
|
||||
}
|
||||
|
||||
/** Produces the one MPEG-TS stream every output copies. */
|
||||
export interface Source {
|
||||
/** Starts producing; each chunk goes to write. */
|
||||
start(config: StreamConfig, write: (chunk: Uint8Array) => void): Promise<void> | void;
|
||||
stop(): void;
|
||||
readonly running: boolean;
|
||||
}
|
||||
|
||||
/** Listens for one RTMP publisher (OBS) and passes its stream on unchanged. */
|
||||
export class RtmpIngest implements Source {
|
||||
private process: Subprocess<'ignore', 'pipe', 'pipe'> | null = null;
|
||||
/** When the current listener started; OBS should connect after it is up. */
|
||||
listeningSince = 0;
|
||||
/** When a publisher last sent data. */
|
||||
receivedAt = 0;
|
||||
|
||||
constructor(
|
||||
private options: { ffmpeg: string; host: string; port: number; key: string; log: Log }
|
||||
) {}
|
||||
|
||||
get running() {
|
||||
return !!this.process && this.process.exitCode == null && !this.process.killed;
|
||||
}
|
||||
|
||||
/** Receiving data from OBS in the last few seconds. */
|
||||
get receiving() {
|
||||
return this.running && Date.now() - this.receivedAt < 5000;
|
||||
}
|
||||
|
||||
get url() {
|
||||
return `rtmp://${this.options.host}:${this.options.port}/live`;
|
||||
}
|
||||
|
||||
start(_config: StreamConfig, write: (chunk: Uint8Array) => void) {
|
||||
if (this.running) return;
|
||||
const { ffmpeg, host, port, key, log } = this.options;
|
||||
const process = Bun.spawn(
|
||||
[
|
||||
ffmpeg,
|
||||
'-hide_banner',
|
||||
'-loglevel',
|
||||
'error',
|
||||
'-listen',
|
||||
'1',
|
||||
'-i',
|
||||
`rtmp://${host}:${port}/live/${key}`,
|
||||
'-c',
|
||||
'copy',
|
||||
'-f',
|
||||
'mpegts',
|
||||
'pipe:1'
|
||||
],
|
||||
{ stdout: 'pipe', stderr: 'pipe', windowsHide: true }
|
||||
);
|
||||
this.process = process;
|
||||
this.listeningSince = Date.now();
|
||||
log(`Waiting for OBS on ${this.url}`);
|
||||
void (async () => {
|
||||
for await (const chunk of process.stdout) {
|
||||
this.receivedAt = Date.now();
|
||||
write(chunk);
|
||||
}
|
||||
})();
|
||||
void (async () => {
|
||||
let last = '';
|
||||
for await (const chunk of process.stderr)
|
||||
last = new TextDecoder().decode(chunk).trim() || last;
|
||||
const code = await process.exited;
|
||||
// OBS stopped or dropped; the next apply() listens again.
|
||||
if (this.process === process) {
|
||||
this.process = null;
|
||||
if (code !== 0 && last) log(`Ingest stopped (${code}): ${last}`);
|
||||
}
|
||||
})();
|
||||
}
|
||||
|
||||
stop() {
|
||||
const process = this.process;
|
||||
this.process = null;
|
||||
process?.kill();
|
||||
}
|
||||
}
|
||||
|
||||
interface Session {
|
||||
id: string;
|
||||
storagePrefix: string | null;
|
||||
startedAt: number;
|
||||
}
|
||||
|
||||
export interface BroadcasterOptions {
|
||||
/** Null streams without Vash managing sessions (no recording). */
|
||||
api: VashApi | null;
|
||||
ffmpeg: string;
|
||||
hlsDir: string;
|
||||
source: Source;
|
||||
log: Log;
|
||||
}
|
||||
|
||||
/** Follows a channel's desired state: on or off air, destinations, recording. */
|
||||
export class Broadcaster {
|
||||
onAir = false;
|
||||
session: Session | null = null;
|
||||
private outputs = new Map<string, Output>();
|
||||
private vod: { name: string; duration: number }[] = [];
|
||||
private uploaded = new Set<string>();
|
||||
private lastVodUpload = 0;
|
||||
private posterDone = false;
|
||||
private lastReport = 0;
|
||||
private config: StreamConfig | null = null;
|
||||
private busy = false;
|
||||
|
||||
constructor(private options: BroadcasterOptions) {}
|
||||
|
||||
private get log() {
|
||||
return this.options.log;
|
||||
}
|
||||
|
||||
/** Reconciles with the latest channel settings. Call every few seconds. */
|
||||
async apply(config: StreamConfig) {
|
||||
if (this.busy) return;
|
||||
this.busy = true;
|
||||
try {
|
||||
this.config = config;
|
||||
if (config.live && !this.onAir) await this.goLive(config);
|
||||
else if (!config.live && this.onAir) await this.goOffAir();
|
||||
else if (this.onAir) {
|
||||
this.syncDestinations(config);
|
||||
if (!this.options.source.running)
|
||||
await this.startSource(config).catch((err) =>
|
||||
this.log('The stream source did not start:', (err as Error).message)
|
||||
);
|
||||
}
|
||||
} finally {
|
||||
this.busy = false;
|
||||
}
|
||||
}
|
||||
|
||||
/** Uploads recording segments and reports destination states. Call every second. */
|
||||
async tick() {
|
||||
if (!this.onAir) return;
|
||||
await this.pumpRecording().catch((err) =>
|
||||
this.log('Recording upload failed:', (err as Error).message)
|
||||
);
|
||||
if (Date.now() - this.lastReport > 10_000) {
|
||||
this.lastReport = Date.now();
|
||||
await this.report();
|
||||
}
|
||||
}
|
||||
|
||||
destinations(): OutputStatus[] {
|
||||
return [...this.outputs.values()]
|
||||
.filter((output) => output.id !== 'recording')
|
||||
.map((output) => output.status());
|
||||
}
|
||||
|
||||
get recording() {
|
||||
return this.outputs.has('recording');
|
||||
}
|
||||
|
||||
private write = (chunk: Uint8Array) => {
|
||||
for (const output of this.outputs.values()) output.write(chunk);
|
||||
};
|
||||
|
||||
private async startSource(config: StreamConfig) {
|
||||
await this.options.source.start(config, this.write);
|
||||
}
|
||||
|
||||
private async goLive(config: StreamConfig) {
|
||||
const { api, hlsDir, ffmpeg } = this.options;
|
||||
if (api) {
|
||||
const started = await api.report<{ id: string; storagePrefix: string | null }>({
|
||||
event: 'start'
|
||||
});
|
||||
this.session = { ...started, startedAt: Date.now() };
|
||||
this.vod.length = 0;
|
||||
this.uploaded.clear();
|
||||
this.posterDone = false;
|
||||
}
|
||||
if (config.record && this.session?.storagePrefix) {
|
||||
await rm(hlsDir, { recursive: true, force: true });
|
||||
await mkdir(hlsDir, { recursive: true });
|
||||
this.outputs.set(
|
||||
'recording',
|
||||
new Output(
|
||||
'recording',
|
||||
'Recording',
|
||||
[
|
||||
'-c',
|
||||
'copy',
|
||||
'-f',
|
||||
'hls',
|
||||
'-hls_time',
|
||||
'2',
|
||||
'-hls_list_size',
|
||||
'8',
|
||||
'-hls_segment_filename',
|
||||
join(hlsDir, 'seg_%05d.ts'),
|
||||
join(hlsDir, 'index.m3u8')
|
||||
],
|
||||
ffmpeg,
|
||||
this.log
|
||||
)
|
||||
);
|
||||
}
|
||||
this.syncDestinations(config);
|
||||
// On air from here: the session exists, so a source that fails to start is
|
||||
// retried by apply() rather than opening another session.
|
||||
this.onAir = true;
|
||||
await this.startSource(config).catch((err) =>
|
||||
this.log('The stream source did not start:', (err as Error).message)
|
||||
);
|
||||
this.log(
|
||||
`On air: ${config.destinations.map((each) => each.label).join(', ') || 'recording only'}`
|
||||
);
|
||||
}
|
||||
|
||||
private async goOffAir() {
|
||||
this.onAir = false;
|
||||
this.options.source.stop();
|
||||
for (const output of this.outputs.values()) output.stop();
|
||||
this.outputs.clear();
|
||||
await Bun.sleep(2500);
|
||||
await this.pumpRecording(true).catch((err) =>
|
||||
this.log('Final upload failed:', (err as Error).message)
|
||||
);
|
||||
const { api } = this.options;
|
||||
if (api && this.session) {
|
||||
await api
|
||||
.report({
|
||||
event: 'end',
|
||||
sessionId: this.session.id,
|
||||
durationSec: Math.round((Date.now() - this.session.startedAt) / 1000)
|
||||
})
|
||||
.catch(() => {});
|
||||
}
|
||||
this.session = null;
|
||||
this.log('Off air');
|
||||
}
|
||||
|
||||
private syncDestinations(config: StreamConfig) {
|
||||
const wanted = new Map(config.destinations.map((destination) => [destination.id, destination]));
|
||||
for (const [id, output] of this.outputs) {
|
||||
if (id === 'recording') continue;
|
||||
const destination = wanted.get(id);
|
||||
if (!destination || output.url !== destination.url) {
|
||||
output.stop();
|
||||
this.outputs.delete(id);
|
||||
this.log(`${output.label}: ${destination ? 'settings changed, reconnecting' : 'removed'}`);
|
||||
}
|
||||
}
|
||||
for (const destination of config.destinations) {
|
||||
if (this.outputs.has(destination.id)) continue;
|
||||
this.outputs.set(
|
||||
destination.id,
|
||||
new Output(
|
||||
destination.id,
|
||||
destination.label,
|
||||
['-c', 'copy', '-f', 'flv', destination.url],
|
||||
this.options.ffmpeg,
|
||||
this.log,
|
||||
destination.url
|
||||
)
|
||||
);
|
||||
this.log(`${destination.label}: connecting`);
|
||||
}
|
||||
}
|
||||
|
||||
private async upload(
|
||||
files: Record<string, { body: Uint8Array | string; type: string; cache?: string }>
|
||||
) {
|
||||
const { api } = this.options;
|
||||
if (!api || !this.session?.storagePrefix) return;
|
||||
const names = Object.keys(files);
|
||||
const { urls } = await api.uploadUrls(this.session.id, names);
|
||||
await Promise.all(
|
||||
names.map(async (name) => {
|
||||
const file = files[name];
|
||||
const response = await fetch(urls[name], {
|
||||
method: 'PUT',
|
||||
body: file.body,
|
||||
headers: {
|
||||
'content-type': file.type,
|
||||
...(file.cache ? { 'cache-control': file.cache } : {})
|
||||
}
|
||||
});
|
||||
if (!response.ok) throw new Error(`upload ${name}: ${response.status}`);
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
private vodPlaylist(ended: boolean) {
|
||||
const target = Math.ceil(Math.max(2, ...this.vod.map((each) => each.duration)));
|
||||
return [
|
||||
'#EXTM3U',
|
||||
'#EXT-X-VERSION:3',
|
||||
`#EXT-X-TARGETDURATION:${target}`,
|
||||
'#EXT-X-MEDIA-SEQUENCE:0',
|
||||
`#EXT-X-PLAYLIST-TYPE:${ended ? 'VOD' : 'EVENT'}`,
|
||||
...this.vod.flatMap((each) => [`#EXTINF:${each.duration.toFixed(3)},`, `live/${each.name}`]),
|
||||
...(ended ? ['#EXT-X-ENDLIST'] : [])
|
||||
].join('\n');
|
||||
}
|
||||
|
||||
/** Uploads new segments, then the live playlist, and now and then the growing VOD playlist. */
|
||||
private async pumpRecording(final = false) {
|
||||
if (!this.session?.storagePrefix) return;
|
||||
const { hlsDir, ffmpeg } = this.options;
|
||||
let playlist: string;
|
||||
try {
|
||||
playlist = await readFile(join(hlsDir, 'index.m3u8'), 'utf8');
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
const entries = parsePlaylist(playlist);
|
||||
const fresh = entries.filter((entry) => !this.uploaded.has(entry.name));
|
||||
if (fresh.length) {
|
||||
const files: Record<string, { body: Uint8Array; type: string }> = {};
|
||||
for (const entry of fresh)
|
||||
files[`live/${entry.name}`] = {
|
||||
body: new Uint8Array(await readFile(join(hlsDir, entry.name))),
|
||||
type: 'video/mp2t'
|
||||
};
|
||||
await this.upload(files);
|
||||
for (const entry of fresh) {
|
||||
this.uploaded.add(entry.name);
|
||||
this.vod.push(entry);
|
||||
}
|
||||
await this.upload({
|
||||
'live/index.m3u8': {
|
||||
body: playlist,
|
||||
type: 'application/vnd.apple.mpegurl',
|
||||
cache: 'no-cache, max-age=1'
|
||||
}
|
||||
});
|
||||
}
|
||||
if (final || Date.now() - this.lastVodUpload > 30_000) {
|
||||
this.lastVodUpload = Date.now();
|
||||
await this.upload({
|
||||
'vod.m3u8': {
|
||||
body: this.vodPlaylist(final),
|
||||
type: 'application/vnd.apple.mpegurl',
|
||||
cache: final ? 'max-age=3600' : 'no-cache'
|
||||
}
|
||||
});
|
||||
}
|
||||
// Local segments that left the live playlist are safely stored; free the disk.
|
||||
const live = new Set(entries.map((entry) => entry.name));
|
||||
for (const name of await readdir(hlsDir))
|
||||
if (name.endsWith('.ts') && this.uploaded.has(name) && !live.has(name))
|
||||
await rm(join(hlsDir, name), { force: true });
|
||||
const session = this.session;
|
||||
if (!this.posterDone && session && Date.now() - session.startedAt > 15_000 && entries.length) {
|
||||
this.posterDone = true;
|
||||
const latest = entries[entries.length - 1].name;
|
||||
const poster = join(hlsDir, 'poster.jpg');
|
||||
const grab = Bun.spawn(
|
||||
[
|
||||
ffmpeg,
|
||||
'-hide_banner',
|
||||
'-loglevel',
|
||||
'error',
|
||||
'-y',
|
||||
'-i',
|
||||
join(hlsDir, latest),
|
||||
'-frames:v',
|
||||
'1',
|
||||
'-vf',
|
||||
'scale=640:-2',
|
||||
poster
|
||||
],
|
||||
{ windowsHide: true }
|
||||
);
|
||||
if ((await grab.exited) === 0)
|
||||
await this.upload({
|
||||
'poster.jpg': { body: new Uint8Array(await readFile(poster)), type: 'image/jpeg' }
|
||||
}).catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
private async report() {
|
||||
const { api } = this.options;
|
||||
if (!api || !this.session) return;
|
||||
await api
|
||||
.report({
|
||||
event: 'status',
|
||||
sessionId: this.session.id,
|
||||
durationSec: Math.round((Date.now() - this.session.startedAt) / 1000),
|
||||
destinations: this.destinations()
|
||||
})
|
||||
.catch((err) => this.log('Status report failed:', (err as Error).message));
|
||||
}
|
||||
|
||||
/** Goes off air (ending the session and finishing the recording) if on air. */
|
||||
async stop() {
|
||||
if (this.onAir) await this.goOffAir();
|
||||
this.options.source.stop();
|
||||
}
|
||||
}
|
||||
|
||||
/** Segment names and durations from an HLS media playlist. */
|
||||
export function parsePlaylist(playlist: string) {
|
||||
const entries: { name: string; duration: number }[] = [];
|
||||
const lines = playlist.split(/\r?\n/);
|
||||
for (let index = 0; index < lines.length; index++) {
|
||||
const match = /^#EXTINF:([\d.]+)/.exec(lines[index]);
|
||||
if (match && lines[index + 1]?.trim())
|
||||
entries.push({ name: lines[index + 1].trim(), duration: Number(match[1]) });
|
||||
}
|
||||
return entries;
|
||||
}
|
||||
Reference in New Issue
Block a user