feat: add streaming demuxer tests and implement dual-stream output piping in streaming service

src/media/mediaController.ts

@@ -17,7 +17,7 @@ import { createMusicPlayer } from "./musicPlayer";
 export interface MediaControllerDependencies {
   isVoiceConnected?: () => boolean;
   isBrowserStreaming?: () => boolean;
-  resolveMediaSource?: (source: string) => Promise<ResolvedMediaSource>;
+  resolveMediaSource?: (source: string, mode?: MediaMode) => Promise<ResolvedMediaSource>;
   musicPlayer?: MusicPlayer;
   screenController?: ScreenShareController;
   onStateChange?: (state: MediaState) => void;
@@ -73,12 +73,17 @@ export class MediaController {
     options: QueueMediaOptions = {},
   ): Promise<MediaState> {
     const mode = options.mode ?? "music";
+
+    const resolved = await (
+      this.dependencies.resolveMediaSource ?? resolveMediaSource
+    )(source, mode);
+
     if (mode === "screen") {
       // Stop current music if any
       this.playbackToken++;
       this.playback?.stop();
       this.playback = null;
-      return this.startScreen(source);
+      return this.startScreen(resolved.source);
     }
 
     // mode === "music"
@@ -95,9 +100,6 @@ export class MediaController {
     }
 
     this.assertCanStartMusic();
-    const resolved = await (
-      this.dependencies.resolveMediaSource ?? resolveMediaSource
-    )(source);
     this.queueStore.add(resolved, mode, options.requestedBy);
     this.startNextIfIdle();
     return this.emitState();

src/media/mediaResolver.ts

@@ -1,7 +1,7 @@
 import { existsSync, statSync } from "node:fs";
 import path from "node:path";
 import { AppError } from "../errors";
-import type { ResolvedMediaSource } from "./mediaTypes";
+import type { ResolvedMediaSource, MediaMode } from "./mediaTypes";
 import { createPlayDlResolver } from "./playDlResolver";
 import { createYtDlp, type YtDlpClient } from "./ytdlp";
 
@@ -18,7 +18,10 @@ export function createMediaResolver(
   const ytdlp = dependencies.ytdlp ?? createYtDlp();
   const playDlResolver = dependencies.playDlResolver ?? createPlayDlResolver();
 
-  return async function resolve(input: string): Promise<ResolvedMediaSource> {
+  return async function resolve(
+    input: string,
+    mode: MediaMode = "music"
+  ): Promise<ResolvedMediaSource> {
     const source = input.trim();
     if (!source) {
       throw new AppError(
@@ -31,13 +34,17 @@ export function createMediaResolver(
     const url = parseUrl(source);
     if (url && isYouTubeUrl(url)) {
       const metadata = await ytdlp.getMetadata(source);
-      const directUrl = await ytdlp.getDirectAudioUrl(source);
+      const directUrl = mode === "screen" 
+        ? await ytdlp.getDirectVideoUrl(source)
+        : await ytdlp.getDirectAudioUrl(source);
       return { source: directUrl, title: metadata.title, kind: "youtube" };
     }
 
     if (url && isSpotifyTrackUrl(url)) {
       const result = await playDlResolver.resolveSpotifyTrack(source);
-      const directUrl = await ytdlp.getDirectAudioUrl(result.url);
+      const directUrl = mode === "screen"
+        ? await ytdlp.getDirectVideoUrl(result.url)
+        : await ytdlp.getDirectAudioUrl(result.url);
       return { source: directUrl, title: result.title, kind: "spotify" };
     }
 
@@ -55,7 +62,9 @@ export function createMediaResolver(
 
     if (!url && !looksLikeUrl(source)) {
       const result = await playDlResolver.searchYouTube(source);
-      const directUrl = await ytdlp.getDirectAudioUrl(result.url);
+      const directUrl = mode === "screen"
+        ? await ytdlp.getDirectVideoUrl(result.url)
+        : await ytdlp.getDirectAudioUrl(result.url);
       return { source: directUrl, title: result.title, kind: "search" };
     }
 

src/media/ytdlp.ts

@@ -56,7 +56,7 @@ export function createYtDlp(dependencies: YtDlpDependencies = {}): YtDlpClient {
         url,
         "--get-url",
         "--format",
-        "bestvideo[protocol^=http]+bestaudio[protocol^=http]/best[protocol^=http]/best",
+        "best[protocol^=http]/best",
         "--no-playlist",
         "--no-warnings",
         "--quiet",

src/streaming/index.ts

@@ -75,6 +75,7 @@ export class Streamer {
         const bitrateStr = String(options.bitrate ?? 2500).replace(/k$/i, "");
         const bitrateVideo = parseInt(bitrateStr, 10) || 2500;
 
+        console.log("[Streamer] Starting screen share for source:", typeof targetSource === "string" ? targetSource.slice(0, 50) + "..." : "ReadableStream");
         const { command, output } = dankPrepareStream(targetSource, {
           encoder: Encoders.software({
             x264: { preset: (options.presetH26x as any) ?? "superfast" },
@@ -86,6 +87,8 @@ export class Streamer {
           bitrateVideo: bitrateVideo,
           frameRate: fps,
           includeAudio: options.includeAudio !== false,
+          minimizeLatency: false,
+          customInputOptions: ["-fflags nobuffer"],
           customHeaders: {
             "User-Agent":
               "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.3",
@@ -95,6 +98,12 @@ export class Streamer {
 
         currentCommand = command;
 
+        const webOutput = new PassThrough();
+        const discordOutput = new PassThrough();
+        
+        output.pipe(webOutput);
+        output.pipe(discordOutput);
+
         const globalAny: any = globalThis;
         const onData = (chunk: Buffer) => {
           try {
@@ -103,21 +112,27 @@ export class Streamer {
             // ignore
           }
         };
-        output.on("data", onData);
+        webOutput.on("data", onData);
 
         command.on("error", (err: Error) => {
-          console.error("Transcoder error:", err);
+          console.error("[Streamer] Transcoder error:", err);
+        });
+        command.on("stderr", (stderrLine: string) => {
+          console.error("[Streamer] FFMPEG:", stderrLine);
+        });
+        command.on("end", () => {
+          console.log("[Streamer] FFMPEG process ended naturally.");
         });
 
         try {
-          await dankPlayStream(output, this.dankStreamer, {
-            type: "go-live",
-            width: 1280,
-            height: 720,
-            frameRate: fps,
-          });
+          console.log("[Streamer] Calling dankPlayStream...");
+          await dankPlayStream(discordOutput, this.dankStreamer, undefined);
+          console.log("[Streamer] dankPlayStream completed successfully.");
+        } catch (err) {
+          console.error("[Streamer] dankPlayStream error:", err);
         } finally {
-          output.off("data", onData);
+          console.log("[Streamer] Cleaning up stream resources.");
+          webOutput.off("data", onData);
           stop();
         }
       },