perf(golive): ffmpeg-spawn demuxer (no node-av) + E2E pipeline tests

Phase 2 — replace the 114MB node-av binary with a plain ffmpeg spawn:

Demuxer.ts: spool stream input to temp file → probe via ffmpeg stderr
(ffmpeg-headless ships NO ffprobe — parse 'Stream #0:0: Video: h264...
640x360, 30 fps' from -loglevel info) → ffmpeg -c copy -f h264 pipe:1
→ NAL-split frames. Falls back to h264 defaults when probe fails.

prepareStream.ts: resolve ffmpeg from FFMPEG_PATH env → Nix store
ffmpeg-headless (hash-prefixed entry!) → PATH; split encoder option
strings ('-forced-idr 1' → two argv) — fluent-ffmpeg used to split
automatically, spawn does not.

E2E tests (tsx, need LD_LIBRARY_PATH=/tmp/ldc-build):
- golive-demux-e2e.ts: real H264 file → 33 NAL frames + dims from probe
- golive-pipeline-e2e.ts: prepareStream → demux → 82 frames
- golive-videostream-e2e.ts: local peer pair → demux → VideoStream →
  native setPacketizer/sendFrame/addTimestamp → 33 frames sent connected

Pitfalls captured: setPacketizer before negotiation breaks createOffer
('No DataChannel or Track to negotiate'); track methods are read-only
(no monkeypatching); both peers must declare audio+video tracks or
answer hangs; state() returns 'closed' after close() — snapshot first.

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;
       };

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: <store>/<hash>-ffmpeg-headless-*/bin/<name>
+  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<Array<Record<string, unknown>>> {
   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<Record<string, unknown>> = [];
+      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<string, unknown> = {
+          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<void>((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<Record<string, unknown>> = [];
-  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 };

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<any>;
   };
 }
@@ -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);
   }
 

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));

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);
+});

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);

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);
+});