diff --git a/services/discord-gateway/src/modules/voice-recording/mediaSource.ts b/services/discord-gateway/src/modules/voice-recording/mediaSource.ts index abf958a..54a3b4b 100644 --- a/services/discord-gateway/src/modules/voice-recording/mediaSource.ts +++ b/services/discord-gateway/src/modules/voice-recording/mediaSource.ts @@ -3,7 +3,9 @@ import { chmodSync, existsSync, mkdtempSync, + readdirSync, rmSync, + statSync, writeFileSync, } from "node:fs"; import { tmpdir } from "node:os"; @@ -478,27 +480,24 @@ export function resolveMediaUrl( } /** - * Resolve a media URL to a single playable input stream for screen share / - * GoLive streaming. + * Download the merged video+audio media for screen share to a TEMP FILE + * first, then return the file path. * - * Streams the merged video+audio media directly from yt-dlp stdout (`-o -`). + * Screen-share playback does NOT stream from yt-dlp stdout anymore: the + * merge feed is delivered at network speed (bursts + stalls), and a live + * pipe defeats ffmpeg's `-re` throttle (the input PTS timeline is + * unreliable), so the encoder force-duplicates held frames → ~1fps video. + * Downloading the FULL clip to a file first gives the encoder a clean, + * monotonic-PTS input, where `-re` (applied in prepareStream for string + * inputs) pacing is proven reliable. * - * 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 Readable of the merged media stream. + * @returns absolute path of the completed media file (caller should delete + * it via cleanup after playback ends). */ -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. +export function downloadScreenInput(url: string): Promise { + return new Promise((resolve, reject) => { + // Temp dir per run (world-writable like /tmp) so parallel/retry runs + // never collide on merge fragments. const tmpDir = mkdtempSync(join(tmpdir(), "gmw-ytdlp-")); chmodSync(tmpDir, 0o1777); @@ -506,18 +505,18 @@ export function getDirectScreenInput(url: string): Promise { const args = [ "-f", "bestvideo[protocol^=http]+bestaudio[protocol^=http]/best[protocol^=http]/best", + // File output (NOT `-o -`): yt-dlp merges DASH fragments into a real + // container with clean timestamps, which is what -re needs to pace. "-o", - "-", + join(tmpDir, "media.%(ext)s"), "--no-playlist", "--no-warnings", "--no-progress", ...cookieArgs, - "-P", - tmpDir, url, ]; - logger.info({ url }, "Spawning yt-dlp for screen share input streaming"); + logger.info({ url }, "Downloading full media for screen share input"); const proc = spawn("yt-dlp", args, { stdio: ["ignore", "pipe", "pipe"], @@ -525,9 +524,6 @@ export function getDirectScreenInput(url: string): Promise { activeProcesses.add(proc); - const stream = new PassThrough(); - proc.stdout.pipe(stream); - let stderrBuf = ""; const MAX_STDERR = 4096; proc.stderr?.on("data", (chunk: Buffer) => { @@ -536,41 +532,58 @@ export function getDirectScreenInput(url: string): Promise { } }); - let producedData = false; - stream.once("data", () => { - producedData = true; - }); + let settled = false; + const failOnce = (message: string) => { + if (settled) return; + settled = true; + activeProcesses.delete(proc); + rmSync(tmpDir, { recursive: true, force: true }); + reject(new Error(message)); + }; proc.on("error", (err: NodeJS.ErrnoException) => { - activeProcesses.delete(proc); - rmSync(tmpDir, { recursive: true, force: true }); - if (err.code === "ENOENT") { - stream.destroy(buildNotInstalledError()); - } else { - stream.destroy(new Error(`yt-dlp failed to start: ${err.message}`)); - } + failOnce(`yt-dlp failed to start: ${err.message}`); }); + let procFinished = false; proc.on("close", (code) => { + procFinished = true; activeProcesses.delete(proc); - 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) { + if (code !== 0) { const detail = stderrBuf.trim() ? `: ${stderrBuf.trim()}` : ""; - stream.destroy( - new Error( - `yt-dlp screen input stream failed (exit ${code})${detail}`, - ), - ); + failOnce(`yt-dlp download failed (exit ${code})${detail}`); + return; + } + // Find the media file yt-dlp wrote (skip .part / .ytdl temp state). + let mediaPath: string | null = null; + for (const entry of readdirSync(tmpDir)) { + if (entry.endsWith(".part") || entry.endsWith(".ytdl")) continue; + mediaPath = join(tmpDir, entry); + break; + } + if (mediaPath && existsSync(mediaPath) && statSync(mediaPath).size > 0) { + settled = true; + resolve(mediaPath); + } else { + failOnce("yt-dlp finished but produced no media file"); } }); - // Resolve immediately — data flows as yt-dlp downloads. The caller's - // resolveInputWithRetry validates the first byte and retries on failure. - resolve(stream); + // Safety net: a stalled download must not hang the gateway forever. + const timer = setTimeout( + () => { + try { + proc.kill("SIGTERM"); + } catch { + /* already dead */ + } + if (!procFinished) { + failOnce("yt-dlp download timed out (10 min)"); + } + }, + 10 * 60 * 1000, + ); + proc.once("close", () => clearTimeout(timer)); }); } diff --git a/services/discord-gateway/src/modules/voice-recording/screenShareController.ts b/services/discord-gateway/src/modules/voice-recording/screenShareController.ts index 644b39a..dddb593 100644 --- a/services/discord-gateway/src/modules/voice-recording/screenShareController.ts +++ b/services/discord-gateway/src/modules/voice-recording/screenShareController.ts @@ -1,4 +1,5 @@ -import { PassThrough, type Readable } from "node:stream"; +import { rmSync } from "node:fs"; +import { dirname } from "node:path"; import type { Client } from "discord.js-selfbot-v13"; import { createChildLogger } from "@/shared/logger/index"; import { @@ -9,7 +10,7 @@ import { Streamer, } from "../../goLive/index.js"; import { - getDirectScreenInput, + downloadScreenInput, INVIDIOUS_INSTANCES, isYoutubeWatchUrl, toInvidiousUrl, @@ -61,17 +62,17 @@ export class ScreenShareController { } /** - * Resolve the screen-share input with retry + first-byte validation. + * Resolve the screen-share input with retry (downloads the FULL media to a + * temp file; returns the file path). * - * Transient YouTube 403s kill the merge ffmpeg BEFORE it produces any - * output; without validation the stream would "start" with a dead input - * and show a black tile forever. So after getDirectScreenInput resolves we - * tee the stream through a PassThrough and wait for the FIRST readable - * byte (or an error / early EOF). On failure the whole resolution is - * retried with a FRESH yt-dlp run (signed DASH URLs expire quickly — the - * old URLs cannot simply be re-fetched). + * Transient YouTube 403s kill a download BEFORE completion; we validate + * the finished file and retry with a FRESH yt-dlp run (signed DASH URLs + * expire quickly). Downloading to a file (instead of streaming the merge + * pipe) gives the encoder a monotonic-PTS input, so ffmpeg `-re` pacing + * in prepareStream actually works (it does NOT on live pipes — the root + * of the ~1fps video). */ - private async resolveInputWithRetry(source: string): Promise { + private async resolveInputWithRetry(source: string): Promise { const MAX_ATTEMPTS = 3; let lastError: Error | null = null; @@ -99,73 +100,12 @@ export class ScreenShareController { } try { - const input = await getDirectScreenInput(source); - - const tee = new PassThrough(); - input.on("error", (err) => tee.destroy(err)); - input.on("end", () => tee.end()); - input.pipe(tee); - // If the merge process is stuck (no data, no exit) destroy the raw - // stream too so ffmpeg gets EPIPE on its next write and dies — - // otherwise every failed attempt leaks a merge process. - const destroyInput = () => { - try { - input.destroy(); - } catch { - /* already gone */ - } - }; - - await new Promise((resolve, reject) => { - const timer = setTimeout(() => { - cleanup(); - destroyInput(); - // Listeners were just removed by cleanup() — destroying tee WITH - // an error would emit "error" on an unlistened PassThrough and - // surface as an unhandled 'error' event (crash). Destroy - // silently; the error lives in the rejection only. - tee.destroy(); - reject( - new Error( - "Screen input produced no data within 12s — merge likely failed", - ), - ); - }, 12000); - const onReadable = () => { - if (tee.readableLength > 0) { - cleanup(); - resolve(); - } - // readableLength === 0 can mean "EOF reached" — handled by onEnd. - }; - const onError = (err: Error) => { - cleanup(); - reject(err); - }; - const onEnd = () => { - cleanup(); - destroyInput(); - reject(new Error("Screen input ended before producing any data")); - }; - const cleanup = () => { - clearTimeout(timer); - tee.removeListener("readable", onReadable); - tee.removeListener("error", onError); - tee.removeListener("end", onEnd); - }; - tee.once("readable", onReadable); - tee.once("error", onError); - tee.once("end", onEnd); - }); - - // Safety net: cleanup() removes the once() listeners on timeout/error, - // but a late error event from input.pipe(tee) can still fire on an - // unlistened PassThrough and crash the gateway (unhandled 'error'). - // A permanent no-op listener guarantees the event is always swallowed. - tee.on("error", () => {}); - // Pass the tee onward — the encoder consumes the same buffered - // stream, so no data from the merge is lost. - return tee; + const mediaPath = await downloadScreenInput(source); + this.logger.info( + { mediaPath, attempt }, + "Screen input downloaded to file", + ); + return mediaPath; } catch (err) { lastError = err instanceof Error ? err : new Error(String(err)); this.logger.warn( @@ -174,7 +114,7 @@ export class ScreenShareController { maxAttempts: MAX_ATTEMPTS, error: lastError.message, }, - "Screen input resolution failed; retrying with fresh yt-dlp", + "Screen input download failed; retrying with fresh yt-dlp", ); if (attempt < MAX_ATTEMPTS) { await new Promise((r) => setTimeout(r, 1500 * attempt)); @@ -250,6 +190,19 @@ export class ScreenShareController { }); const { command } = prepared; + // The downloaded temp media lives in a per-run tmpdir; remove it once + // playback is done (natural end, failure, or user stop). The tmpdir is + // the parent of the media file, so deleting it removes the file too. + const cleanupTempMedia = () => { + try { + if (typeof input === "string") { + rmSync(dirname(input), { recursive: true, force: true }); + } + } catch { + /* best-effort — tmp dirs are world-writable, leak is bounded */ + } + }; + let stopped = false; // Restore the @discordjs/voice connection after the stream ends (both // natural end and failure), so the user can keep using audio/mic. @@ -262,6 +215,7 @@ export class ScreenShareController { /* already dead */ } } + cleanupTempMedia(); try { this.streamer?.voiceConnection?.stop(); } catch { @@ -312,6 +266,7 @@ export class ScreenShareController { } catch { /* already dead */ } + cleanupTempMedia(); // Leave the voice channel the Streamer joined (its own connection). try { this.streamer?.voiceConnection?.stop(); diff --git a/services/discord-gateway/tests/screenShareInput.test.ts b/services/discord-gateway/tests/screenShareInput.test.ts index a502cf9..1d9176d 100644 --- a/services/discord-gateway/tests/screenShareInput.test.ts +++ b/services/discord-gateway/tests/screenShareInput.test.ts @@ -1,12 +1,14 @@ // ═══════════════════════════════════════════════════════════════════════════════ // Screen share input resolution tests // -// 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. +// downloadScreenInput downloads the FULL merged video+audio media to a temp +// file first (yt-dlp `-o /media.%(ext)s`), then returns the file +// path. Feeding a FILE path (not a live stdout pipe) to prepareStream is +// what makes ffmpeg `-re` pacing reliable — a pipe has unreliable PTS and +// caused the ~1fps force-duplication symptom. // -// yt-dlp is faked via a PATH shim script 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 { @@ -18,34 +20,41 @@ import { } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { Readable } from "node:stream"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { getDirectScreenInput } from "../src/modules/voice-recording/mediaSource.js"; +import { downloadScreenInput } from "../src/modules/voice-recording/mediaSource.js"; // ─── fake bin dir ────────────────────────────────────────────────────────────── let fakeBinDir: string | null = null; const realPath = process.env.PATH; -beforeAll(() => { - fakeBinDir = mkdtempSync(join(tmpdir(), "gmw-fake-bins-")); - - // Fake yt-dlp: streams a few bytes to stdout (like `yt-dlp -o -` does). - // Modes (env): - // GMW_FAKE_YTDLP_FAIL=1 → stderr 403 + exit 8 WITHOUT stdout bytes - // (mimics a download rejected by YouTube). - const ytShim = `#!/usr/bin/env bash +// Bash shim: find the -o pattern, substitute the extension, write 4096 bytes. +// Escaped as a separate string to keep the TS template literal simple. +const ytShimBody = `prev="" +out="" +for a in "$@"; do + if [ "$prev" = "-o" ]; then out="$a"; fi + prev="$a" +done if [ "$GMW_FAKE_YTDLP_FAIL" = "1" ]; then echo "ERROR: [youtube] ...: 403 Forbidden (access denied)" >&2 exit 8 fi -# Fake yt-dlp — ignore args, emit a few bytes so consumers see a live stream. -head -c 4096 /dev/urandom +target=$(printf '%s' "$out" | sed 's/%(ext)s/.mp4/') +head -c 4096 /dev/urandom > "$target" exit 0 `; +const ytShim = `#!/usr/bin/env bash\n${ytShimBody}`; + +const ytShimDump = `#!/usr/bin/env bash +printf '%s\\n' "$*" >> "$GMW_FAKE_YTDLP_DUMP_ARGS" +${ytShimBody} +`; + +beforeAll(() => { + fakeBinDir = mkdtempSync(join(tmpdir(), "gmw-fake-bins-")); writeFileSync(join(fakeBinDir, "yt-dlp"), ytShim); chmodSync(join(fakeBinDir, "yt-dlp"), 0o755); - process.env.PATH = `${fakeBinDir}:${process.env.PATH}`; }); @@ -56,72 +65,47 @@ afterAll(() => { process.env.PATH = realPath; }); -// ─── helpers ─────────────────────────────────────────────────────────────────── -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 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); - - 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$/); +describe("downloadScreenInput", () => { + it("downloads media to a temp file and returns its path", async () => { + const mediaPath = await downloadScreenInput("https://youtu.be/abc"); + expect(typeof mediaPath).toBe("string"); + expect(mediaPath).toMatch(/gmw-ytdlp-/); + const size = readFileSync(mediaPath).length; + expect(size).toBeGreaterThan(0); + // The fake yt-dlp writes 4096 bytes → the file must deliver them. + expect(size).toBe(4096); }); - 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. + it("rejects when yt-dlp fails (transient 403) so the controller retries", async () => { 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 consumeStream(result); - expect(outcome).toMatch(/^(error|end)-after-0B$/); + await expect(downloadScreenInput("https://youtu.be/abc")).rejects.toThrow( + /403|exit 8|failed/i, + ); } finally { delete process.env.GMW_FAKE_YTDLP_FAIL; } }); - it("passes -o - (stdout streaming) and a temp dir to yt-dlp", async () => { + it("passes a file -o pattern (NOT `-o -`) to yt-dlp", async () => { const argsDump = join( tmpdir(), `gmw-ytargs-${process.pid}-${Date.now()}.txt`, ); 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); + writeFileSync(existing, ytShimDump); chmodSync(existing, 0o755); try { - const result = await getDirectScreenInput("https://youtu.be/abc"); - await consumeStream(result); + const mediaPath = await downloadScreenInput("https://youtu.be/abc"); + expect(readFileSync(mediaPath).length).toBeGreaterThan(0); await new Promise((r) => setTimeout(r, 100)); const args = readFileSync(argsDump, "utf8").trim(); - expect(args).toContain("-o -"); + expect(args).not.toContain("-o -"); + expect(args).toContain("-o "); expect(args).toMatch(/gmw-ytdlp-/); } finally { delete process.env.GMW_FAKE_YTDLP_DUMP_ARGS;