diff --git a/services/discord-gateway/src/goLive/BaseMediaConnection.ts b/services/discord-gateway/src/goLive/BaseMediaConnection.ts index c80a5b1..b568d6a 100644 --- a/services/discord-gateway/src/goLive/BaseMediaConnection.ts +++ b/services/discord-gateway/src/goLive/BaseMediaConnection.ts @@ -323,7 +323,7 @@ a=ice-lite } const { op, d, seq } = JSON.parse(e.data as string) as { op: number; - // eslint-disable-next-line @typescript-eslint/no-explicit-any -- Discord voice WS payload is dynamically typed + // biome-ignore lint/suspicious/noExplicitAny: Discord voice WS payload is dynamically typed d: any; seq?: number; }; diff --git a/services/discord-gateway/src/goLive/Demuxer.ts b/services/discord-gateway/src/goLive/Demuxer.ts index a4b4b55..919b1e6 100644 --- a/services/discord-gateway/src/goLive/Demuxer.ts +++ b/services/discord-gateway/src/goLive/Demuxer.ts @@ -14,17 +14,44 @@ import { spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; +import { createWriteStream, existsSync, readdirSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { PassThrough } from "node:stream"; -export enum AVCodecID { - AV_CODEC_ID_H264 = 27, - AV_CODEC_ID_HEVC = 173, - AV_CODEC_ID_VP8 = 139, - AV_CODEC_ID_VP9 = 167, - AV_CODEC_ID_AV1 = 225, - AV_CODEC_ID_OPUS = 86019, +/** + * Resolve ffmpeg/ffprobe binary. Prefers explicit env override, then PATH, + * then a Nix-store ffmpeg-headless (the GMW flake provides it in the service + * profile, but dev shells / tests may not have it on PATH). + */ +function resolveBin(name: "ffmpeg"): string { + const override = process.env.FFMPEG_PATH; + if (override && existsSync(override)) return override; + // Nix store scan: /-ffmpeg-headless-*/bin/ + const store = "/nix/store"; + if (existsSync(store)) { + const entries = readdirSync(store); + for (const entry of entries) { + if (!entry.includes("ffmpeg-headless-")) continue; + const candidate = join(store, entry, "bin", name); + if (existsSync(candidate)) return candidate; + } + } + return name; // fall back to PATH } +const FFMPEG = resolveBin("ffmpeg"); + +export const AVCodecID = { + AV_CODEC_ID_H264: 27, + AV_CODEC_ID_HEVC: 173, + AV_CODEC_ID_VP8: 139, + AV_CODEC_ID_VP9: 167, + AV_CODEC_ID_AV1: 225, + AV_CODEC_ID_OPUS: 86019, +} as const; +export type AVCodecID = (typeof AVCodecID)[keyof typeof AVCodecID]; + export const AV_PKT_FLAG_KEY = 1; export interface Frame { @@ -48,37 +75,66 @@ export interface DemuxedStream { stream: PassThrough; } -/** Run ffprobe JSON on a file URL, return raw stream descriptors. */ +/** + * Probe a media file for stream info using ffmpeg's stderr (the + * ffmpeg-headless Nix package ships ffmpeg but not ffprobe). Returns + * stream descriptors in the same shape ffprobe -show_streams would. + */ export async function probeStreams( url: string, ): Promise>> { return new Promise((resolve, reject) => { - const proc = spawn("ffprobe", [ + const proc = spawn(FFMPEG, [ "-hide_banner", "-loglevel", - "error", + "info", "-i", url, - "-print_format", - "json", - "-show_streams", + "-f", + "null", + "-", ]); - let stdout = ""; let stderr = ""; - proc.stdout.on("data", (d: Buffer) => (stdout += d.toString())); proc.stderr.on("data", (d: Buffer) => (stderr += d.toString())); - proc.on("close", (code) => { - if (code === 0) { - try { - const parsed = JSON.parse(stdout); - resolve(parsed.streams ?? []); - } catch (e) { - reject(new Error(`Failed to parse ffprobe output: ${e}`)); + proc.on("close", () => { + // Parse "Stream #0:0: Video: h264 (High), yuv420p, 640x360, 30 fps" + const streams: Array> = []; + const re = /Stream #0:(\d+): (Video|Audio): ([^,]+)/g; + let m: RegExpExecArray | null; + // biome-ignore lint/suspicious/noAssignInExpressions: regex loop idiom + while ((m = re.exec(stderr)) !== null) { + const [full, idx, kind, codecRaw] = m; + void full; + const codecName = codecRaw.split(" ")[0].toLowerCase(); + const stream: Record = { + index: Number(idx), + codec_type: kind.toLowerCase(), + codec_name: codecName, + width: 0, + height: 0, + r_frame_rate: "0/1", + sample_rate: 0, + }; + // dimensions: "640x360" + const dim = /(\d{2,5})x(\d{2,5})/.exec(stderr.slice(m.index)); + if (dim) { + stream.width = Number(dim[1]); + stream.height = Number(dim[2]); } - } else { - reject(new Error(`ffprobe failed (${code}): ${stderr}`)); + // fps: "30 fps" or "29.97 fps" + const fps = /(\d+(?:\.\d+)?) fps/.exec(stderr.slice(m.index)); + if (fps) { + const v = Number(fps[1]); + stream.r_frame_rate = `${Math.round(v * 1000)}/1000`; + } + // sample rate for audio: "48000 Hz" + const sr = /(\d+) Hz/.exec(stderr.slice(m.index)); + if (sr) stream.sample_rate = Number(sr[1]); + streams.push(stream); } + resolve(streams); }); + proc.on("error", (err) => reject(err)); }); } @@ -100,15 +156,44 @@ export async function demux( const vPipe = new PassThrough({ objectMode: true, highWaterMark: 128 }); const aPipe = new PassThrough({ objectMode: true, highWaterMark: 128 }); + // For stream input, spool to a temp file first so ffprobe can inspect it + // (ffprobe needs a seekable file; pipes can't be re-read). The stream is + // fully consumed before ffmpeg starts — acceptable for screen-share + // sources which are already fully buffered by yt-dlp in practice. + let spoolPath: string | null = null; + const cleanupSpool = () => { + if (spoolPath) { + import("node:fs").then(({ unlink }) => unlink(spoolPath!, () => {})); + spoolPath = null; + } + }; + + let effectiveInput: string; + if (typeof input === "string") { + effectiveInput = input; + } else { + spoolPath = join(tmpdir(), `golive-demux-${_label}.h264`); + const ws = createWriteStream(spoolPath); + await new Promise((resolve, reject) => { + input.pipe(ws); + input.on("error", reject); + ws.on("finish", resolve); + ws.on("error", reject); + }); + effectiveInput = spoolPath; + } + // Probe for codec + dimensions let streams: Array> = []; - if (typeof input === "string") { - streams = await probeStreams(input); + try { + streams = await probeStreams(effectiveInput); + } catch (_e) { + // probe failed (e.g. raw h264 without container) — infer h264 default + streams = []; } const v = streams.find((s) => s.codec_type === "video"); const a = streams.find((s) => s.codec_type === "audio"); - let vInfo: DemuxedStream | undefined; let aInfo: DemuxedStream | undefined; @@ -121,7 +206,7 @@ export async function demux( AVCodecID[ (codecName.toUpperCase() as keyof typeof AVCodecID) ?? "AV_CODEC_ID_H264" - ], + ] ?? AVCodecID.AV_CODEC_ID_H264, codecName, width: (v.width as number) ?? 0, height: (v.height as number) ?? 0, @@ -130,6 +215,19 @@ export async function demux( sample_rate: 0, stream: vPipe, }; + } else { + // Probe failed (e.g. raw AnnexB h264 input) — still emit frames on the + // video pipe; playStream infers dimensions from the first frame. + vInfo = { + codec: AVCodecID.AV_CODEC_ID_H264, + codecName: "h264", + width: 0, + height: 0, + framerate_num: 0, + framerate_den: 1, + sample_rate: 0, + stream: vPipe, + }; } if (a) { @@ -151,12 +249,12 @@ export async function demux( } // Spawn ffmpeg — extract raw video (AnnexB for H264) to stdout - const isUrl = typeof input === "string"; const args: string[] = [ "-hide_banner", "-loglevel", "error", - ...(isUrl ? ["-i", input] : ["-i", "pipe:0"]), + "-i", + effectiveInput, "-c:v", "copy", "-an", // no audio in this minimal demuxer @@ -165,15 +263,7 @@ export async function demux( "pipe:1", ]; - const proc = isUrl - ? spawn("ffmpeg", args, { stdio: ["ignore", "pipe", "pipe"] }) - : spawn("ffmpeg", args, { stdio: ["pipe", "pipe", "pipe"] }); - - if (proc.stdin && !isUrl) { - input.on("data", (chunk: Buffer) => proc.stdin?.write(chunk)); - input.on("end", () => proc.stdin?.end()); - input.on("error", () => proc.stdin?.destroy()); - } + const proc = spawn(FFMPEG, args, { stdio: ["ignore", "pipe", "pipe"] }); // Scan stdout for NAL units. Each NAL unit (between start codes) is one frame // payload. We emit them individually; the packetizer chain handles FU-A. @@ -280,6 +370,7 @@ export async function demux( proc.kill("SIGTERM"); vPipe.end(); aPipe.end(); + cleanupSpool(); }; return { video: vInfo, audio: aInfo, close }; diff --git a/services/discord-gateway/src/goLive/Streamer.ts b/services/discord-gateway/src/goLive/Streamer.ts index db1198b..e86d5c4 100644 --- a/services/discord-gateway/src/goLive/Streamer.ts +++ b/services/discord-gateway/src/goLive/Streamer.ts @@ -27,7 +27,7 @@ export interface StreamerClientLike { broadcast(data: { op: number; d: unknown }): void; }; guilds?: { - // eslint-disable-next-line @typescript-eslint/no-explicit-any -- discord.js-selfbot client shape is dynamic + // biome-ignore lint/suspicious/noExplicitAny: discord.js-selfbot client shape is dynamic fetch(id: string): Promise; }; } @@ -199,7 +199,7 @@ export class Streamer { const { guildId } = this.voiceConnection.streamConnection; if (!this.client.guilds) return; const server = await this.client.guilds.fetch(guildId); - // eslint-disable-next-line @typescript-eslint/no-unsafe-call, @typescript-eslint/no-explicit-any -- discord.js-selfbot dynamic + // biome-ignore lint/suspicious/noExplicitAny: discord.js-selfbot dynamic (server as any).members.me?.voice?.postPreview(data); } diff --git a/services/discord-gateway/src/goLive/prepareStream.ts b/services/discord-gateway/src/goLive/prepareStream.ts index c140dc7..41a50ee 100644 --- a/services/discord-gateway/src/goLive/prepareStream.ts +++ b/services/discord-gateway/src/goLive/prepareStream.ts @@ -9,6 +9,8 @@ */ import { type ChildProcess, spawn } from "node:child_process"; +import { existsSync, readdirSync } from "node:fs"; +import { join } from "node:path"; import { PassThrough, type Readable } from "node:stream"; import { demux } from "./Demuxer.js"; import { type EncoderSettings, Encoders } from "./Encoders.js"; @@ -37,6 +39,25 @@ const DEFAULT_HEADERS = { Connection: "keep-alive", }; +/** Resolve ffmpeg binary (env override → PATH → Nix store ffmpeg-headless). */ +function resolveFfmpeg(): string { + if (process.env.FFMPEG_PATH && existsSync(process.env.FFMPEG_PATH)) { + return process.env.FFMPEG_PATH; + } + const store = "/nix/store"; + if (existsSync(store)) { + const entries = readdirSync(store); + for (const entry of entries) { + if (!entry.includes("ffmpeg-headless-")) continue; + const candidate = join(store, entry, "bin", "ffmpeg"); + if (existsSync(candidate)) return candidate; + } + } + return "ffmpeg"; +} + +const FFMPEG_BIN = resolveFfmpeg(); + /** * prepareStream — build an ffmpeg command (as spawn args + PassThrough output) * that transcodes the input into a pipe we can demux. Mirrors @dank074's @@ -132,6 +153,11 @@ export function prepareStream( throw new Error( `Encoder settings not specified for ${mergedOptions.videoCodec}`, ); + // Encoder options are declared as single strings like "-forced-idr 1"; + // spawn needs each flag and value as separate argv entries. + const encOptions = enc.options.flatMap((opt) => + opt.split(/\s+/).filter(Boolean), + ); args.push( "-b:v", `${mergedOptions.bitrateVideo}k`, @@ -147,8 +173,10 @@ export function prepareStream( "expr:gte(t,n_forced*1)", "-c:v", enc.name, - ...enc.options, - ...(enc.globalOptions ?? []), + ...encOptions, + ...(enc.globalOptions ?? []).flatMap((opt) => + opt.split(/\s+/).filter(Boolean), + ), ); } @@ -174,8 +202,8 @@ export function prepareStream( const isUrl = typeof input === "string"; const proc: ChildProcess = isUrl - ? spawn("ffmpeg", args, { stdio: ["ignore", "pipe", "pipe"] }) - : spawn("ffmpeg", args, { stdio: ["pipe", "pipe", "pipe"] }); + ? spawn(FFMPEG_BIN, args, { stdio: ["ignore", "pipe", "pipe"] }) + : spawn(FFMPEG_BIN, args, { stdio: ["pipe", "pipe", "pipe"] }); if (proc.stdin && !isUrl) { input.on("data", (chunk: Buffer) => proc.stdin?.write(chunk)); diff --git a/services/discord-gateway/tests/golive-demux-e2e.ts b/services/discord-gateway/tests/golive-demux-e2e.ts new file mode 100644 index 0000000..56f49aa --- /dev/null +++ b/services/discord-gateway/tests/golive-demux-e2e.ts @@ -0,0 +1,40 @@ +// Phase 2 E2E: Demuxer on a real ffmpeg-generated H264 file. +// Run: npx tsx tests/golive-demux-e2e.ts + +import { createReadStream } from "node:fs"; +import { demux } from "../src/goLive/Demuxer.js"; + +const input = process.argv[2] ?? "/tmp/sample.h264"; +const { video, close } = await demux(createReadStream(input), { + format: "h264", +}); + +console.log( + "video:", + JSON.stringify({ + codecName: video.codecName, + width: video.width, + height: video.height, + duration: video.duration, + fps: Math.round(video.framerate_num / video.framerate_den), + }), +); + +let count = 0; +let keyframes = 0; +let bytes = 0; +video.stream.on("data", (frame: { data: Buffer; keyframe: boolean }) => { + count++; + bytes += frame.data.length; + if (frame.keyframe) keyframes++; +}); +video.stream.on("end", () => { + console.log(`frames: ${count} (${keyframes} keyframes), ${bytes} bytes`); + close(); + process.exit(0); +}); +video.stream.on("error", (e: unknown) => { + console.error("stream error:", e); + close(); + process.exit(1); +}); diff --git a/services/discord-gateway/tests/golive-pipeline-e2e.ts b/services/discord-gateway/tests/golive-pipeline-e2e.ts new file mode 100644 index 0000000..042006a --- /dev/null +++ b/services/discord-gateway/tests/golive-pipeline-e2e.ts @@ -0,0 +1,58 @@ +// Phase 2 E2E: full pipeline prepareStream → demux → frame stream. +// Run: npx tsx tests/golive-pipeline-e2e.ts + +import { demux } from "../src/goLive/Demuxer.js"; +import { Encoders } from "../src/goLive/Encoders.js"; +import { prepareStream } from "../src/goLive/prepareStream.js"; +import { normalizeVideoCodec } from "../src/goLive/utils.js"; + +// Use a real ffmpeg-generated video file as input (from sample generation). +const input = process.argv[2] ?? "/tmp/sample.h264"; + +const prepared = prepareStream(input, { + encoder: Encoders.software({ x264: { preset: "superfast" } }), + width: 640, + height: 360, + frameRate: 25, + bitrateVideo: 500, + bitrateVideoMax: 800, + includeAudio: false, + videoCodec: normalizeVideoCodec("H264"), +}); + +console.log( + "prepareStream ok, videoCodec:", + prepared.videoCodec, + "size:", + prepared.width, + "x", + prepared.height, +); + +const { video, close } = await demux(prepared.output, { format: "h264" }); +console.log("demux video:", video?.codecName, video?.width, "x", video?.height); + +let frames = 0; +let keyframes = 0; +video.stream.on("data", (f: { keyframe?: boolean }) => { + frames++; + if (f.keyframe) keyframes++; +}); +video.stream.on("end", () => { + console.log(`pipeline frames: ${frames} (${keyframes} keyframes)`); + close(); + prepared.command.kill("SIGTERM"); + process.exit(frames > 0 ? 0 : 1); +}); +video.stream.on("error", (e: unknown) => { + console.error("pipeline error:", e); + close(); + prepared.command.kill("SIGTERM"); + process.exit(1); +}); +setTimeout(() => { + console.log("timeout after 30s — killing"); + close(); + prepared.command.kill("SIGTERM"); + process.exit(2); +}, 30000); diff --git a/services/discord-gateway/tests/golive-videostream-e2e.ts b/services/discord-gateway/tests/golive-videostream-e2e.ts new file mode 100644 index 0000000..4cb7b68 --- /dev/null +++ b/services/discord-gateway/tests/golive-videostream-e2e.ts @@ -0,0 +1,78 @@ +// Phase 2 E2E: demux → VideoStream → native packetizer chain (local pair). +// Run: npx tsx tests/golive-videostream-e2e.ts + +import { createReadStream } from "node:fs"; +import { demux } from "../src/goLive/Demuxer.js"; +import { loadNative } from "../src/goLive/native.js"; +import { VideoStream } from "../src/goLive/VideoStream.js"; + +async function main() { + const native = loadNative(); + const { PeerConnection } = native; + + const pcA = new PeerConnection({ iceServers: [] }); + const pcB = new PeerConnection({ iceServers: [] }); + + pcA.onStateChange(() => {}); + pcB.onStateChange(() => {}); + + // Both peers declare audio+video tracks (exact passing test-packetizer + // pattern — tracks trigger negotiation). + pcA.addTrack("0", "audio"); + pcA.addTrack("1", "video"); + pcB.addTrack("0", "audio"); + const trackB = pcB.addTrack("1", "video"); + if (!trackB) throw new Error("no track from addTrack"); + + const track = trackB; + // NOTE: setPacketizer is called AFTER connected (see below) — calling it + // before negotiation breaks the offer (libdatachannel negotiation state). + + const offer = await pcA.createOffer(); + console.log("T1 offer"); + pcB.setRemoteDescription(offer, "offer"); + const answer = await pcB.createAnswer(offer); + console.log("T2 answer"); + pcA.setRemoteDescription(answer, "answer"); + + await new Promise((r) => setTimeout(r, 1500)); + console.log("T3 states:", pcA.state(), "/", pcB.state()); + + // Discord-style SSRC/payload: H264 101 @ 90kHz, playout ext id 5 + track.setPacketizer("h264", 0x1234, 101, 90000, 5, 0, 10); + + const { video, close } = await demux(createReadStream("/tmp/sample.h264"), { + format: "h264", + }); + console.log("video stream:", video.codecName, video.width, "x", video.height); + + const conn = { + sendVideoFrame: (frame: Buffer, frametime: number) => { + track.sendFrame(frame); + track.addTimestamp(Math.round((frametime * 90000) / 1000)); + }, + } as unknown as { sendVideoFrame(frame: Buffer, frametime: number): void }; + + const vStream = new VideoStream(conn as never); + let sent = 0; + const origSend = conn.sendVideoFrame; + conn.sendVideoFrame = (frame: Buffer, frametime: number) => { + sent++; + origSend(frame, frametime); + }; + + video.stream.pipe(vStream); + await new Promise((r) => setTimeout(r, 4000)); + + console.log(`sent ${sent} frames via VideoStream; B state=${pcB.state()}`); + const ok = sent > 0 && pcB.state() === "connected"; + close(); + pcA.close(); + pcB.close(); + process.exit(ok ? 0 : 1); +} + +main().catch((e) => { + console.error("E2E failed:", e); + process.exit(1); +});