diff --git a/services/discord-gateway/src/modules/voice-recording/mediaSource.ts b/services/discord-gateway/src/modules/voice-recording/mediaSource.ts index ffd275d..8529ca5 100644 --- a/services/discord-gateway/src/modules/voice-recording/mediaSource.ts +++ b/services/discord-gateway/src/modules/voice-recording/mediaSource.ts @@ -1,4 +1,7 @@ import { type ChildProcess, spawn } from "node:child_process"; +import { chmodSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { PassThrough, type Readable } from "node:stream"; import { StreamType } from "@discordjs/voice"; import { createChildLogger } from "@/shared/logger/index"; @@ -378,283 +381,97 @@ export function resolveMediaUrl( * Resolve a media URL to a single playable input stream for screen share / * GoLive streaming. * - * yt-dlp `--get-url` with `bestvideo+bestaudio` prints the video-only and - * audio-only URLs on SEPARATE lines. The old code took only the first line - * (video-only) → ffmpeg had no audio track → GoLive stream had no sound. + * Streams the merged video+audio media directly from yt-dlp stdout (`-o -`). * - * This returns a single input that `prepareStream` (which accepts only ONE - * ffmpeg input) can consume while STILL including audio: - * - If yt-dlp offers a merged progressive URL (one URL, video+audio) it is - * returned directly. - * - Otherwise the video-only + audio-only DASH URLs are fetched in the SAME - * yt-dlp run (signature URLs expire quickly) and merged locally by an - * ffmpeg process into a single NUT stream, which is streamed to the - * consumer over a Readable. NUT over stdin auto-probes cleanly (verified: - * av1+opus merge → H264+opus transcode). + * This is deliberately NOT the old --dump-single-json + manual URL-fetch + * approach: YouTube signs DASH URLs for the extracting client and rejects + * them with 403 when fetched raw by ffmpeg/curl (verified 2026-08-12: even + * curl with the EXACT http_headers from the yt-dlp dump got 403 on some + * videos, while yt-dlp's own downloader succeeded). Streaming from yt-dlp + * lets it handle auth, cookies and transient retries internally — the same + * mechanism resolveMediaUrl already uses for music playback. * - * @returns a direct video URL (string) or a Readable of the merged NUT stream. + * @returns a Readable of the merged media stream. */ -export function getDirectScreenInput(url: string): Promise { - return new Promise((resolve, reject) => { +export function getDirectScreenInput(url: string): Promise { + return new Promise((resolve) => { + // Merge fragments must NOT be written to the process CWD — the Nix + // store dir is read-only for the deployed gateway (EACCES). Use a + // per-run temp dir (world-writable like /tmp) so parallel/retry runs + // never collide on merge fragments and any user can write to it. + const tmpDir = mkdtempSync(join(tmpdir(), "gmw-ytdlp-")); + chmodSync(tmpDir, 0o1777); + const args = [ - url, - "--dump-single-json", - "--format", + "-f", "bestvideo[protocol^=http]+bestaudio[protocol^=http]/best[protocol^=http]/best", + "-o", + "-", "--no-playlist", "--no-warnings", - "--quiet", - // NOTE: deliberately NOT --no-simulate. Simulate mode still resolves the - // requested format URLs into the JSON (requested_formats[].url), and it - // avoids yt-dlp writing .part files into the process CWD — which is the - // read-only Nix store dir for the deployed gateway (EACCES). + "--no-progress", + "-P", + tmpDir, + url, ]; - logger.info({ url }, "Spawning yt-dlp for screen share input resolution"); + logger.info({ url }, "Spawning yt-dlp for screen share input streaming"); const proc = spawn("yt-dlp", args, { - stdio: ["pipe", "pipe", "pipe"], + stdio: ["ignore", "pipe", "pipe"], }); activeProcesses.add(proc); - let stdoutBuf = ""; + const stream = new PassThrough(); + proc.stdout.pipe(stream); + let stderrBuf = ""; const MAX_STDERR = 4096; - const MAX_STDOUT = 8 * 1024 * 1024; // JSON metadata + requested format URLs + proc.stderr?.on("data", (chunk: Buffer) => { + if (stderrBuf.length < MAX_STDERR) { + stderrBuf += chunk.toString("utf8"); + } + }); - if (proc.stdout) { - proc.stdout.on("data", (chunk: Buffer) => { - if (stdoutBuf.length < MAX_STDOUT) { - stdoutBuf += chunk - .toString("utf8") - .slice(0, MAX_STDOUT - stdoutBuf.length); - } - }); - } - - if (proc.stderr) { - proc.stderr.on("data", (chunk: Buffer) => { - if (stderrBuf.length < MAX_STDERR) { - stderrBuf += chunk - .toString("utf8") - .slice(0, MAX_STDERR - stderrBuf.length); - } - }); - } + let producedData = false; + stream.once("data", () => { + producedData = true; + }); proc.on("error", (err: NodeJS.ErrnoException) => { activeProcesses.delete(proc); + rmSync(tmpDir, { recursive: true, force: true }); if (err.code === "ENOENT") { - reject(buildNotInstalledError()); + stream.destroy(buildNotInstalledError()); } else { - reject(new Error(`yt-dlp failed to start: ${err.message}`)); + stream.destroy(new Error(`yt-dlp failed to start: ${err.message}`)); } }); proc.on("close", (code) => { activeProcesses.delete(proc); - - if (code !== 0) { + rmSync(tmpDir, { recursive: true, force: true }); + // Fail fast: a download that dies before producing ANY bytes (e.g. + // transient YouTube 403) cannot feed the encoder — destroy the stream + // so the caller retries with a fresh yt-dlp run instead of streaming + // a silent black tile. + if (code !== 0 && !producedData && !stream.destroyed) { const detail = stderrBuf.trim() ? `: ${stderrBuf.trim()}` : ""; - reject( + stream.destroy( new Error( - `yt-dlp screen input resolution exited with code ${code}${detail}`, + `yt-dlp screen input stream failed (exit ${code})${detail}`, ), ); - return; } - - let parsed: Record; - try { - parsed = JSON.parse(stdoutBuf.trim()) as Record; - } catch (parseErr) { - reject( - new Error( - `Failed to parse yt-dlp JSON for screen input: ${(parseErr as Error).message}`, - ), - ); - return; - } - - resolveScreenInput(parsed).then(resolve, (err: unknown) => { - const message = err instanceof Error ? err.message : String(err); - reject( - new Error(`Failed to build screen input for "${url}": ${message}`), - ); - }); }); + + // Resolve immediately — data flows as yt-dlp downloads. The caller's + // resolveInputWithRetry validates the first byte and retries on failure. + resolve(stream); }); } -/** - * From a parsed yt-dlp JSON info dict, decide how to feed a single ffmpeg - * input with both video and audio. - */ -async function resolveScreenInput( - info: Record, -): Promise { - const requested = info.requested_formats as - | Array> - | undefined; - - // Merged/progressive single URL (video+audio in one). Common when yt-dlp - // selects a single format (e.g. format 18 progressive mp4) or when a direct - // muxed URL is available. - const singleUrl = info.url as string | undefined; - const singleHasAudio = - info.acodec !== "none" && - typeof info.acodec === "string" && - info.acodec.length > 0; - - if (typeof singleUrl === "string" && singleUrl && singleHasAudio) { - logger.debug("Screen share uses merged progressive single URL"); - return singleUrl; - } - - // Separate video-only + audio-only DASH formats → merge locally via ffmpeg. - if (Array.isArray(requested) && requested.length >= 2) { - const video = requested.find( - (rf) => rf.vcodec && String(rf.vcodec) !== "none", - ); - const audio = requested.find( - (rf) => rf.acodec && String(rf.acodec) !== "none", - ); - const videoUrl = video?.url as string | undefined; - const audioUrl = audio?.url as string | undefined; - // yt-dlp returns the exact HTTP headers needed to fetch each signed DASH - // URL (User-Agent etc.). Passing them to the merge ffmpeg prevents - // transient YouTube 403s ("Server returned 403 Forbidden") that kill the - // stream before it starts. - const videoHeaders = - (video?.http_headers as Record | undefined) ?? {}; - - if ( - typeof videoUrl === "string" && - videoUrl.length > 0 && - typeof audioUrl === "string" && - audioUrl.length > 0 - ) { - return mergeScreenStreams(videoUrl, audioUrl, videoHeaders); - } - } - - throw new Error( - "yt-dlp returned neither a merged progressive URL nor a video+audio format pair", - ); -} - -/** - * Merge a video-only URL and an audio-only URL into a single NUT stream using - * a child ffmpeg process. Both URLs come from the same yt-dlp run, so they - * share the same signature/expiry and are consumed immediately. - * - * Fail-fast contract: if the merge process exits non-zero BEFORE producing any - * output bytes (e.g. transient YouTube 403), the returned Readable is - * destroyed with an error so the caller can retry — otherwise the screen - * share would "start" with a dead input and stream a black tile forever. - */ -function mergeScreenStreams( - videoUrl: string, - audioUrl: string, - httpHeaders?: Record, -): Readable { - logger.info("Merging video+audio DASH streams into a single NUT input"); - - const args = [ - "-hide_banner", - "-loglevel", - "error", - "-reconnect", - "1", - "-reconnect_streamed", - "1", - "-reconnect_delay_max", - "5", - ]; - // Pass the browser-like headers yt-dlp attached to the signed URLs. Without - // a proper User-Agent YouTube sometimes answers 403 to ffmpeg's plain - // Lavf/… agent and the whole stream dies before producing a frame. - const headerStr = Object.entries(httpHeaders ?? {}) - .map(([k, v]) => `${k}: ${v}`) - .join("\r\n"); - if (headerStr) { - args.push("-headers", headerStr); - } - args.push( - "-i", - videoUrl, - "-i", - audioUrl, - "-map", - "0:v:0", - "-map", - "1:a:0", - "-c:v", - "copy", - "-c:a", - "copy", - "-f", - "nut", - "pipe:1", - ); - - const ffmpeg = spawn("ffmpeg", args, { - stdio: ["ignore", "pipe", "pipe"], - }); - - // Track so cleanup() can terminate the merge during graceful shutdown. - activeProcesses.add(ffmpeg); - ffmpeg.once("exit", () => { - activeProcesses.delete(ffmpeg); - }); - - // Prevent the ffmpeg stderr from filling the pipe buffer / leaking. - let stderrBuf = ""; - const MAX_STDERR = 4096; - ffmpeg.stderr?.on("data", (chunk: Buffer) => { - if (stderrBuf.length < MAX_STDERR) { - stderrBuf += chunk.toString("utf8"); - } - }); - - let producedData = false; - - ffmpeg.on("error", (err) => { - const msg = - err.message === "spawn ffmpeg ENOENT" - ? "FFmpeg not found! Install ffmpeg in the container." - : err.message; - logger.error({ error: msg }, "Screen stream merge ffmpeg error"); - stream.destroy(new Error(msg)); - }); - - ffmpeg.on("exit", (code) => { - const stderr = stderrBuf.trim(); - logger.warn( - { code, stderr: stderr.slice(-500) || undefined }, - "Screen stream merge ffmpeg exited", - ); - // Fail fast: a merge that dies before emitting ANY bytes cannot feed the - // encoder — destroy the stream with an error so the caller retries with a - // fresh resolution instead of streaming a silent black tile. - if (code !== 0 && !producedData && !stream.destroyed) { - stream.destroy( - new Error( - `Screen stream merge failed before producing data (exit ${code})${stderr ? `: ${stderr.slice(-300)}` : ""}`, - ), - ); - } - }); - - const stream = ffmpeg.stdout; - stream.setMaxListeners(32); - stream.once("data", () => { - producedData = true; - }); - return stream; -} - /** * Extract metadata (title, duration, thumbnail) from a media URL * without downloading the audio stream. diff --git a/services/discord-gateway/src/modules/voice-recording/screenShareController.ts b/services/discord-gateway/src/modules/voice-recording/screenShareController.ts index 768b5d8..2a03524 100644 --- a/services/discord-gateway/src/modules/voice-recording/screenShareController.ts +++ b/services/discord-gateway/src/modules/voice-recording/screenShareController.ts @@ -66,20 +66,13 @@ export class ScreenShareController { * retried with a FRESH yt-dlp run (signed DASH URLs expire quickly — the * old URLs cannot simply be re-fetched). */ - private async resolveInputWithRetry( - source: string, - ): Promise { + private async resolveInputWithRetry(source: string): Promise { const MAX_ATTEMPTS = 3; let lastError: Error | null = null; for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) { try { const input = await getDirectScreenInput(source); - if (typeof input === "string") { - // Direct URL input — nothing to validate; the encoder ffmpeg will - // connect itself and fail loudly on a bad URL. - return input; - } const tee = new PassThrough(); input.on("error", (err) => tee.destroy(err)); diff --git a/services/discord-gateway/tests/screenShareInput.test.ts b/services/discord-gateway/tests/screenShareInput.test.ts index 59942d6..a502cf9 100644 --- a/services/discord-gateway/tests/screenShareInput.test.ts +++ b/services/discord-gateway/tests/screenShareInput.test.ts @@ -1,13 +1,12 @@ // ═══════════════════════════════════════════════════════════════════════════════ // Screen share input resolution tests // -// Verifies the decision logic of getDirectScreenInput: -// - merged progressive URL → returned directly -// - video+audio DASH pair → local ffmpeg merge (Readable) -// - neither → rejection +// getDirectScreenInput now streams the merged video+audio media straight from +// yt-dlp stdout (`-o -`) — same auth-handling mechanism as resolveMediaUrl for +// music. There is no manual URL fetch or local ffmpeg merge anymore. // -// Both yt-dlp and ffmpeg are faked via PATH shim scripts so the test does not -// hit the network or need real binaries. +// yt-dlp is faked via a PATH shim script so the test does not hit the network +// or need real binaries. // ═══════════════════════════════════════════════════════════════════════════════ import { @@ -31,39 +30,21 @@ const realPath = process.env.PATH; beforeAll(() => { fakeBinDir = mkdtempSync(join(tmpdir(), "gmw-fake-bins-")); - // Fake yt-dlp: prints the JSON file named in GMW_FAKE_YTDLP_JSON. - // If the file is missing → exits 1 (mimics yt-dlp failure). - const ytShim = `#!/usr/bin/env bash -if [ -n "$GMW_FAKE_YTDLP_JSON" ] && [ -f "$GMW_FAKE_YTDLP_JSON" ]; then - cat "$GMW_FAKE_YTDLP_JSON" - exit 0 -fi -echo "yt-dlp: fake JSON missing" >&2 -exit 1 -`; - writeFileSync(join(fakeBinDir, "yt-dlp"), ytShim); - chmodSync(join(fakeBinDir, "yt-dlp"), 0o755); - - // Fake ffmpeg: writes a small nut-ish payload to stdout so the returned - // Readable actually emits data (the merge path in mergeScreenStreams). + // Fake yt-dlp: streams a few bytes to stdout (like `yt-dlp -o -` does). // Modes (env): - // GMW_FAKE_FFMPEG_FAIL=1 → exit 1, no stdout (mimics transient 403) - // GMW_FAKE_FFMPEG_DUMP_ARGS= → append argv to the file (asserts - // flags like -headers are forwarded to the merge process) - const ffShim = `#!/usr/bin/env bash -if [ -n "$GMW_FAKE_FFMPEG_DUMP_ARGS" ]; then - printf '%s\\n' "$*" >> "$GMW_FAKE_FFMPEG_DUMP_ARGS" -fi -if [ "$GMW_FAKE_FFMPEG_FAIL" = "1" ]; then - echo "403 Forbidden" >&2 + // GMW_FAKE_YTDLP_FAIL=1 → stderr 403 + exit 8 WITHOUT stdout bytes + // (mimics a download rejected by YouTube). + const ytShim = `#!/usr/bin/env bash +if [ "$GMW_FAKE_YTDLP_FAIL" = "1" ]; then + echo "ERROR: [youtube] ...: 403 Forbidden (access denied)" >&2 exit 8 fi -# Fake ffmpeg — ignore args, emit a few bytes so consumers see a live stream. +# Fake yt-dlp — ignore args, emit a few bytes so consumers see a live stream. head -c 4096 /dev/urandom exit 0 `; - writeFileSync(join(fakeBinDir, "ffmpeg"), ffShim); - chmodSync(join(fakeBinDir, "ffmpeg"), 0o755); + writeFileSync(join(fakeBinDir, "yt-dlp"), ytShim); + chmodSync(join(fakeBinDir, "yt-dlp"), 0o755); process.env.PATH = `${fakeBinDir}:${process.env.PATH}`; }); @@ -76,161 +57,76 @@ afterAll(() => { }); // ─── helpers ─────────────────────────────────────────────────────────────────── -function writeFakeJson(payload: Record): string { - const p = join( - tmpdir(), - `gmw-fake-${process.pid}-${Date.now()}-${Math.random().toString(36).slice(2)}.json`, - ); - writeFileSync(p, JSON.stringify(payload)); - return p; -} - -function dashPairInfo(videoUrl: string, audioUrl: string) { - return { - url: null, - acodec: "none", // top-level is not a single merged format - vcodec: "av01", - requested_formats: [ - { - format_id: "136", - vcodec: "avc1.4d401f", - acodec: "none", - url: videoUrl, - }, - { format_id: "140", vcodec: "none", acodec: "mp4a.40.2", url: audioUrl }, - ], - }; +function consumeStream(stream: Readable): Promise { + return new Promise((resolve) => { + let got = 0; + stream.on("data", (chunk: Buffer) => { + got += chunk.length; + }); + stream.on("error", () => resolve(`error-after-${got}B`)); + stream.on("end", () => resolve(`end-after-${got}B`)); + stream.resume(); + }); } // ─── tests ───────────────────────────────────────────────────────────────────── describe("getDirectScreenInput", () => { - it("returns the single merged progressive URL when the info has one", async () => { - process.env.GMW_FAKE_YTDLP_JSON = writeFakeJson({ - url: "https://cdn.example/progressive.mp4", - acodec: "mp4a.40.2", - vcodec: "avc1", - }); - const result = await getDirectScreenInput("https://youtu.be/abc"); - expect(result).toBe("https://cdn.example/progressive.mp4"); - }); - - it("returns a live Readable when a video+audio DASH pair must be merged", async () => { - process.env.GMW_FAKE_YTDLP_JSON = writeFakeJson( - dashPairInfo( - "https://cdn.example/video.mp4", - "https://cdn.example/audio.m4a", - ), - ); + it("returns a live Readable and streams media bytes from yt-dlp stdout", async () => { const result = await getDirectScreenInput("https://youtu.be/abc"); expect(Readable.isReadable(result)).toBe(true); - // The fake ffmpeg emits bytes; collect a chunk to prove the stream flows. - const bytes = await new Promise((resolve, reject) => { - const stream = result as Readable; - let got = 0; - stream.on("data", (chunk: Buffer) => { - got += chunk.length; - }); - stream.on("error", reject); - stream.on("end", () => resolve(got)); - stream.resume(); - }); - expect(bytes).toBeGreaterThan(0); + const outcome = await consumeStream(result); + // The fake yt-dlp emits 4096 bytes → the stream must deliver them. + expect(outcome).toMatch(/^(error|end)-after-[1-9]\d*B$/); }); - it("rejects when yt-dlp returns neither a merged URL nor a format pair", async () => { - process.env.GMW_FAKE_YTDLP_JSON = writeFakeJson({ - url: null, - acodec: "none", - vcodec: "none", - requested_formats: [], - }); - await expect(getDirectScreenInput("https://youtu.be/abc")).rejects.toThrow( - /neither a merged progressive URL nor a video\+audio/, - ); - }); - - it("rejects when yt-dlp exits non-zero", async () => { - process.env.GMW_FAKE_YTDLP_JSON = "/nonexistent/gmw-fake.json"; - await expect(getDirectScreenInput("https://youtu.be/abc")).rejects.toThrow( - /screen input resolution exited with code 1/, - ); - }); - - it("terminates with ZERO bytes when the merge ffmpeg fails before producing data (transient 403)", async () => { - // Simulate the 11:50 production failure: yt-dlp resolves fine, but the - // merge ffmpeg hits a transient YouTube 403 and exits non-zero WITHOUT - // emitting a single byte. getDirectScreenInput still resolves (the - // Readable exists) — the fail-fast contract: the stream must terminate - // (error OR end — the end-before-exit ordering makes both possible) - // without ever delivering a frame to a consumer. The controller's - // resolveInputWithRetry turns either signal into a fresh retry. - process.env.GMW_FAKE_YTDLP_JSON = writeFakeJson( - dashPairInfo( - "https://cdn.example/video.mp4", - "https://cdn.example/audio.m4a", - ), - ); - process.env.GMW_FAKE_FFMPEG_FAIL = "1"; + it("destroys the stream with an error when yt-dlp fails before producing data (transient 403)", async () => { + // Simulate the production failure: yt-dlp's downloader hits a transient + // YouTube 403 and exits non-zero WITHOUT emitting a single byte. The + // returned Readable must terminate with zero bytes (error OR end) so the + // controller's resolveInputWithRetry retries with a fresh run instead of + // streaming a silent black tile. + process.env.GMW_FAKE_YTDLP_FAIL = "1"; try { const result = await getDirectScreenInput("https://youtu.be/abc"); expect(Readable.isReadable(result)).toBe(true); - const outcome = await new Promise((resolve) => { - const stream = result as Readable; - let got = 0; - stream.on("data", (chunk: Buffer) => { - got += chunk.length; - }); - stream.on("error", () => resolve(`error-after-${got}B`)); - stream.on("end", () => resolve(`end-after-${got}B`)); - stream.resume(); - }); - // Fail-fast: the consumer must NOT receive any bytes (no black-tile - // zombie stream). Either a destroyed-with-error stream or a clean - // end-before-exit is a valid terminal state — the caller retries. + const outcome = await consumeStream(result); expect(outcome).toMatch(/^(error|end)-after-0B$/); } finally { - delete process.env.GMW_FAKE_FFMPEG_FAIL; + delete process.env.GMW_FAKE_YTDLP_FAIL; } }); - it("forwards yt-dlp http_headers to the merge ffmpeg (-headers)", async () => { - const info = dashPairInfo( - "https://cdn.example/video.mp4", - "https://cdn.example/audio.m4a", - ); - // Add the browser-like headers yt-dlp attaches to signed DASH URLs. - (info.requested_formats[0] as Record).http_headers = { - "User-Agent": - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", - Referer: "https://www.youtube.com/", - }; + it("passes -o - (stdout streaming) and a temp dir to yt-dlp", async () => { const argsDump = join( tmpdir(), - `gmw-ffargs-${process.pid}-${Date.now()}.txt`, + `gmw-ytargs-${process.pid}-${Date.now()}.txt`, ); - process.env.GMW_FAKE_YTDLP_JSON = writeFakeJson(info); - process.env.GMW_FAKE_FFMPEG_DUMP_ARGS = argsDump; + process.env.GMW_FAKE_YTDLP_DUMP_ARGS = argsDump; + // Augment the fake to dump its argv. + const shim = `#!/usr/bin/env bash +printf '%s\\n' "$*" >> "$GMW_FAKE_YTDLP_DUMP_ARGS" +head -c 4096 /dev/urandom +exit 0 +`; + const realPath2 = process.env.PATH; + const dir = fakeBinDir as unknown as string; + const existing = join(dir, "yt-dlp"); + // Overwrite with the argv-dumping variant. + writeFileSync(existing, shim); + chmodSync(existing, 0o755); try { const result = await getDirectScreenInput("https://youtu.be/abc"); - // Consume the stream so the merge ffmpeg process runs to completion. - await new Promise((resolve) => { - const stream = result as Readable; - stream.on("data", () => {}); - stream.on("error", () => resolve()); - stream.on("end", () => resolve()); - stream.resume(); - }); - // Allow the fake ffmpeg to flush its argv dump. + await consumeStream(result); await new Promise((r) => setTimeout(r, 100)); const args = readFileSync(argsDump, "utf8").trim(); - expect(args).toContain("-headers"); - expect(args).toContain("Mozilla/5.0"); - expect(args).toContain("Referer: https://www.youtube.com/"); + expect(args).toContain("-o -"); + expect(args).toMatch(/gmw-ytdlp-/); } finally { - delete process.env.GMW_FAKE_FFMPEG_DUMP_ARGS; + delete process.env.GMW_FAKE_YTDLP_DUMP_ARGS; rmSync(argsDump, { force: true }); + process.env.PATH = realPath2; } }); });