glass beadian
aa5e262f3a
Self-hosted multi-platform broadcast newsroom for activist collectives: publishing pipeline with group approval and DST-proof scheduling, TweetDeck-style column wall with hostile-content-safe rendering, live control room with Auto Director, bridged chat governance with polls and quorum, distributed sandboxed transcode workers, encrypted content-addressed archive, monthly analytics. MIT licensed. See README.md and docs/.
250 lines
11 KiB
TypeScript
250 lines
11 KiB
TypeScript
/**
|
||
* Playlist feeder (streaming-architecture.md §5.1): a `kind: "playlist"` source
|
||
* becomes a continuous input like any other. The feeder walks the ordered asset
|
||
* list, opens each asset (a presigned GET through the ingest egress relay),
|
||
* remuxes it to MPEG-TS (pre-normalized at upload: copy only, never a
|
||
* transcode) and pipes the remuxes one after another into a FIFO that one
|
||
* `playlistPipe` ingest reads at native rate. FFmpeg reads only pipes (WP9);
|
||
* no concat protocol.
|
||
*
|
||
* - Timestamps: each remux is shifted to where the previous asset ended (its
|
||
* `-progress` end time), so the concatenated stream is monotonic. FFmpeg
|
||
* 5.1's own discontinuity correction stalls `-re` for 1–2 s per boundary.
|
||
* - Boundaries: two overlapped stages. Asset i+1's remux starts as soon as
|
||
* asset i's remux has exited (its end time is then known), while asset i is
|
||
* still being fed from its capture buffer (~4 MiB ahead, several seconds), so
|
||
* the next asset's bytes are waiting when the current one ends: no gap.
|
||
* - Failure: an asset that cannot be opened or remuxed is skipped. When a whole
|
||
* pass of the list fails (or the list is empty), the slate floor plays for a
|
||
* few seconds and the list is retried.
|
||
* - End: `loop: false` plays the list once, then the slate for good.
|
||
* - Desired-state v2: `update()` adopts a new asset list without a restart. The
|
||
* asset on air finishes; playback continues after it in the new list (or at
|
||
* the new list's start when it is no longer there). An already prepared next
|
||
* asset that is no longer next is discarded.
|
||
*/
|
||
import { closeSync, constants, openSync, rmSync } from "node:fs";
|
||
import { Socket } from "node:net";
|
||
import { PassThrough, type Readable } from "node:stream";
|
||
import { makeFifo, openBlockingReader } from "../gate/fifo";
|
||
import { spawnManaged, type ManagedProcess, type StderrRedactor } from "../relay/ffmpeg";
|
||
import { buildPlaylistRemuxArgs, buildSlateArgs, type Canvas } from "./command-builder";
|
||
|
||
/** Opens one playlist asset's bytes (WP7: presigned GET through the pinned ingest egress relay). */
|
||
export interface PlaylistAssetSource { open(assetId: string): Promise<Readable> }
|
||
|
||
export interface PlaylistSpec { assetIds: string[]; loop: boolean }
|
||
|
||
export type FeederEvent =
|
||
| { type: "asset"; assetId: string; index: number }
|
||
| { type: "asset-failed"; assetId: string; reason: string }
|
||
| { type: "slate"; reason: "unavailable" | "ended" }
|
||
| { type: "adopted"; assetIds: number };
|
||
|
||
export interface PlaylistFeederDeps {
|
||
assets: PlaylistAssetSource;
|
||
canvas: Canvas;
|
||
scratchDir: string;
|
||
redactor: StderrRedactor;
|
||
onStderr(message: string): void;
|
||
onEvent?(event: FeederEvent): void;
|
||
/** Rendered slate PNG (S4, overlay renderer); null = black. */
|
||
slateImage?: string | null;
|
||
slateSeconds?: number;
|
||
}
|
||
|
||
/** Bytes of produced MPEG-TS held ahead of the FIFO per producer (~5 s at 6 Mb/s). */
|
||
const CAPTURE_BYTES = 4 << 20;
|
||
|
||
/** One remux (or slate) run: its captured output and, once it exits, its end time. */
|
||
interface Job {
|
||
proc: ManagedProcess;
|
||
output: Readable;
|
||
/** Resolves when the producer exits: success, and the end time it reported (µs, excluding the offset). */
|
||
done: Promise<{ ok: boolean; endUs: number; reason: string }>;
|
||
input: Readable | null;
|
||
cancelled: boolean;
|
||
}
|
||
|
||
let nextFeeder = 1;
|
||
|
||
export class PlaylistFeeder {
|
||
private spec: PlaylistSpec;
|
||
private out: Socket | null = null;
|
||
private reader: { fd: number; releaseHolder(): void } | null = null;
|
||
private stopping = false;
|
||
private running: Promise<void> = Promise.resolve();
|
||
private current: { assetId: string; index: number } | null = null;
|
||
private nextIndex = 0;
|
||
/** Bumped by `update()`: a job prepared under an older list is discarded. */
|
||
private generation = 0;
|
||
private readonly jobs = new Set<Job>();
|
||
private abortFeed: (() => void) | null = null;
|
||
/** Assets started, assets failed, slate chunks fed. */
|
||
readonly stats = { assets: 0, failures: 0, slates: 0 };
|
||
ended = false;
|
||
|
||
constructor(readonly sourceId: string, spec: PlaylistSpec, private readonly deps: PlaylistFeederDeps) {
|
||
this.spec = { assetIds: [...spec.assetIds], loop: spec.loop };
|
||
}
|
||
|
||
get nowPlaying(): string | null { return this.current?.assetId ?? null; }
|
||
|
||
/**
|
||
* Creates the FIFO and starts feeding. Returns the blocking read end for the ingest's stdin;
|
||
* call `releaseReader()` once the ingest has it.
|
||
*/
|
||
start(): number {
|
||
const path = makeFifo(this.deps.scratchDir, `playlist-${nextFeeder++}.fifo`);
|
||
const reader = openBlockingReader(path);
|
||
this.reader = reader;
|
||
const out = new Socket({ fd: openSync(path, constants.O_WRONLY | constants.O_NONBLOCK), readable: false, writable: true });
|
||
out.on("error", () => { if (!this.stopping) void this.stop(); }); // the ingest is gone; IngestProcess reports its exit
|
||
this.out = out;
|
||
reader.releaseHolder();
|
||
rmSync(path, { force: true });
|
||
this.running = this.run().catch((error: unknown) => this.deps.onStderr(`playlist feeder ${this.sourceId}: ${String(error)}`));
|
||
return reader.fd;
|
||
}
|
||
|
||
releaseReader(): void {
|
||
if (this.reader) { closeSync(this.reader.fd); this.reader = null; }
|
||
}
|
||
|
||
/** Desired-state v2: adopt a new list at the next asset boundary (no restart, no mid-asset cut). */
|
||
update(spec: PlaylistSpec): void {
|
||
this.spec = { assetIds: [...spec.assetIds], loop: spec.loop };
|
||
const at = this.current ? this.spec.assetIds.indexOf(this.current.assetId) : -1;
|
||
this.nextIndex = at >= 0 ? at + 1 : 0;
|
||
this.ended = false;
|
||
this.generation += 1;
|
||
this.deps.onEvent?.({ type: "adopted", assetIds: this.spec.assetIds.length });
|
||
}
|
||
|
||
private peekNext(): { assetId: string; index: number } | null {
|
||
const ids = this.spec.assetIds;
|
||
if (ids.length === 0) return null;
|
||
let index = this.nextIndex;
|
||
if (index >= ids.length) {
|
||
if (!this.spec.loop) return null;
|
||
index = 0;
|
||
}
|
||
return { assetId: ids[index]!, index };
|
||
}
|
||
|
||
/** Spawns a remux of `asset` (or the slate) at `offsetUs`, capturing its output from the start. */
|
||
private produce(asset: { assetId: string; index: number } | null, offsetUs: number): Job {
|
||
const args = asset ? buildPlaylistRemuxArgs(offsetUs)
|
||
: buildSlateArgs({ canvas: this.deps.canvas, seconds: this.deps.slateSeconds ?? 5, image: this.deps.slateImage ?? null, offsetUs });
|
||
const proc = spawnManaged("ffmpeg", args, [asset ? "pipe" : "ignore", "pipe", "pipe", "pipe"], this.deps.redactor, this.deps.onStderr);
|
||
// A producer may exit before its turn (a broken asset at once, a small one that fits in the
|
||
// buffers), and Node then resumes its unconsumed stdout itself, dropping data and 'end'. The
|
||
// capture keeps both until fed; backpressure holds a large asset's remux CAPTURE_BYTES ahead.
|
||
const output = new PassThrough({ highWaterMark: CAPTURE_BYTES });
|
||
const stdout = proc.child.stdout as Readable | null;
|
||
if (stdout) stdout.pipe(output);
|
||
else output.end();
|
||
let progress = "";
|
||
(proc.child.stdio[3] as Readable | null)?.on("data", (chunk: Buffer) => { progress = (progress + chunk.toString("latin1")).slice(-4096); });
|
||
let failed: string | null = null;
|
||
const done = proc.exited.then(({ code }) => {
|
||
const endUs = Number([...progress.matchAll(/out_time_us=(\d+)/g)].at(-1)?.[1] ?? 0);
|
||
return { ok: failed === null && code === 0 && endUs > 0, endUs, reason: failed ?? `producer exited code=${code} end=${endUs}us` };
|
||
});
|
||
const job: Job = { proc, output, input: null, done, cancelled: false };
|
||
if (asset) {
|
||
proc.child.stdin?.on("error", () => undefined); // EPIPE when the remux rejects its input
|
||
void this.deps.assets.open(asset.assetId).then((input) => {
|
||
if (job.cancelled || this.stopping || proc.child.exitCode !== null || proc.child.signalCode !== null) { input.destroy(); return; }
|
||
job.input = input;
|
||
input.on("error", (error: Error) => { failed = `asset read failed: ${error.name}`; proc.kill("SIGKILL"); });
|
||
if (proc.child.stdin) input.pipe(proc.child.stdin);
|
||
}, (error: unknown) => {
|
||
failed = `asset open failed: ${error instanceof Error ? error.name : "error"}`;
|
||
proc.kill("SIGKILL");
|
||
});
|
||
}
|
||
this.jobs.add(job);
|
||
void done.then(() => this.jobs.delete(job));
|
||
return job;
|
||
}
|
||
|
||
private discard(job: Job): void {
|
||
job.cancelled = true;
|
||
job.input?.destroy();
|
||
job.output.destroy();
|
||
job.proc.kill("SIGKILL");
|
||
}
|
||
|
||
/** Pipes a job's captured output into the FIFO; resolves once all of it has been handed over. */
|
||
private async feed(job: Job): Promise<void> {
|
||
const out = this.out;
|
||
if (!out) return;
|
||
job.output.pipe(out, { end: false });
|
||
// Until the producer's last byte ('end'). On stop the FIFO socket is destroyed, which unpipes
|
||
// and pauses the capture, so 'end' would never come: stop() aborts the wait instead.
|
||
await new Promise<void>((resolve) => {
|
||
this.abortFeed = resolve;
|
||
job.output.once("end", resolve);
|
||
job.output.once("error", () => resolve());
|
||
});
|
||
this.abortFeed = null;
|
||
job.output.unpipe(out);
|
||
}
|
||
|
||
private async run(): Promise<void> {
|
||
let offsetUs = 0;
|
||
let failedInARow = 0;
|
||
let slateNext = false;
|
||
let feeding: Promise<void> = Promise.resolve();
|
||
while (!this.stopping) {
|
||
const generation = this.generation;
|
||
const next = slateNext ? null : this.peekNext();
|
||
slateNext = false;
|
||
// Prepared while the previous job is still being fed (its producer has exited: offsetUs is final).
|
||
const job = this.produce(next, offsetUs);
|
||
await feeding;
|
||
if (this.stopping) { this.discard(job); break; }
|
||
if (generation !== this.generation) { this.discard(job); continue; } // re-plan under the new list
|
||
if (next) {
|
||
this.nextIndex = next.index + 1;
|
||
this.current = next;
|
||
this.stats.assets += 1;
|
||
this.deps.onEvent?.({ type: "asset", assetId: next.assetId, index: next.index });
|
||
} else {
|
||
this.current = null;
|
||
this.ended = this.spec.assetIds.length > 0 && !this.spec.loop && this.nextIndex >= this.spec.assetIds.length;
|
||
this.stats.slates += 1;
|
||
this.deps.onEvent?.({ type: "slate", reason: this.ended ? "ended" : "unavailable" });
|
||
}
|
||
feeding = this.feed(job);
|
||
const result = await job.done;
|
||
if (result.ok) {
|
||
offsetUs += result.endUs;
|
||
failedInARow = 0;
|
||
continue;
|
||
}
|
||
if (!next) continue; // a failed slate is retried (offset unchanged)
|
||
this.stats.failures += 1;
|
||
failedInARow += 1;
|
||
this.deps.onEvent?.({ type: "asset-failed", assetId: next.assetId, reason: result.reason });
|
||
if (failedInARow >= Math.max(1, this.spec.assetIds.length)) {
|
||
failedInARow = 0;
|
||
slateNext = true;
|
||
}
|
||
}
|
||
}
|
||
|
||
async stop(): Promise<void> {
|
||
if (this.stopping) { await this.running; return; }
|
||
this.stopping = true;
|
||
const jobs = [...this.jobs];
|
||
for (const job of jobs) this.discard(job);
|
||
this.abortFeed?.();
|
||
this.out?.destroy();
|
||
this.out = null;
|
||
this.releaseReader();
|
||
await this.running;
|
||
await Promise.all(jobs.map((job) => job.done));
|
||
}
|
||
}
|