fix: YouTube throttling

Fix YouTube video downloading throttling by downloading stream in chunks

.gitignore

@@ -1,6 +1,7 @@
 # npm
 node_modules
 package-lock.json
+pnpm-lock.yaml
 
 # secrets
 .env

src/modules/processing/services/youtube.js

@@ -90,7 +90,10 @@ export default async function(o) {
     let video = adaptive_formats.find(i => ((Number(quality) > Number(bestQuality)) ? checkBestVideo(i) : checkRightVideo(i)));
     if (video && audio) return {
         type: "render",
-        urls: [video.url, audio.url],
+        urls: [
+            video.url + '&cpn=' + info.cpn + '&__clen=' + video.content_length, 
+            audio.url + '&cpn=' + info.cpn + '&__clen=' + audio.content_length
+        ],
         filename: `youtube_${o.id}_${video.width}x${video.height}_${o.format}${isDubbed ? `_${o.dubLang}`:''}.${c[o.format].container}`
     };
 

src/modules/stream/types.js

@@ -1,8 +1,10 @@
 import { spawn } from "child_process";
+import { Constants as YTConstants } from "youtubei.js";
 import ffmpeg from "ffmpeg-static";
 import got from "got";
 import { ffmpegArgs, genericUserAgent } from "../config.js";
 import { metadataManager, msToTime } from "../sub/utils.js";
+import { Readable } from "stream";
 
 export function streamDefault(streamInfo, res) {
     try {
@@ -28,17 +30,79 @@ export function streamDefault(streamInfo, res) {
         res.destroy();
     }
 }
+function createYoutubeStream(format_url) {
+    // We need to download in chunks.
+    const chunk_size = 1048576 * 10; // 10MB
+
+    let chunk_start = 0;
+    let chunk_end = chunk_size;
+    let must_end = false;
+
+    let cancel;
+    let is_fetching = false;
+
+    const url = new URL(format_url);
+    const content_length = Number(url.searchParams.get('__clen'));
+    url.searchParams.delete('__clen');
+    format_url = url.toString();
+
+    return new Readable({
+        read() {
+            if (is_fetching) return;
+            if (must_end) {
+                this.push(null);
+                return;
+            }
+            is_fetching = true;
+            
+            if (chunk_end >= content_length) {
+                must_end = true;
+            }
+            
+            cancel = new AbortController();
+
+            const chunk = got.get(`${format_url}&range=${chunk_start}-${chunk_end || ''}`, {
+                headers: {
+                    ...YTConstants.STREAM_HEADERS
+                },
+                isStream: true,
+                responseType: 'buffer',
+                signal: cancel.signal
+            });
+            
+            chunk.on('data', data => {
+                this.push(data);
+            });
+
+            chunk.on('end', () => {
+                is_fetching = false;
+                chunk_start = chunk_end + 1;
+                chunk_end += chunk_size;
+            });
+            
+            chunk.on('error', e => {
+                this.emit('error', e);
+                chunk.destroy();
+            });
+        },
+        destroy() {
+            if (is_fetching)
+                cancel.abort();
+        }
+    });
+}
 export function streamLiveRender(streamInfo, res) {
     try {
         if (streamInfo.urls.length !== 2) {
             res.destroy();
             return;
         }
-        let audio = got.get(streamInfo.urls[1], { isStream: true });
+        let audio = streamInfo.service === 'youtube' ? createYoutubeStream(streamInfo.urls[1]) : got.get(streamInfo.urls[1], { isStream: true });
+        let video = streamInfo.service === 'youtube' ? createYoutubeStream(streamInfo.urls[0]) : got.get(streamInfo.urls[0], { isStream: true });
 
         let format = streamInfo.filename.split('.')[streamInfo.filename.split('.').length - 1], args = [
             '-loglevel', '-8',
-            '-i', streamInfo.urls[0],
+            '-i', 'pipe:5',
             '-i', 'pipe:3',
             '-map', '0:v',
             '-map', '1:a',
@@ -50,7 +114,7 @@ export function streamLiveRender(streamInfo, res) {
             windowsHide: true,
             stdio: [
                 'inherit', 'inherit', 'inherit',
-                'pipe', 'pipe'
+                'pipe', 'pipe', 'pipe'
             ],
         });
         res.setHeader('Connection', 'keep-alive');
@@ -63,11 +127,11 @@ export function streamLiveRender(streamInfo, res) {
             ffmpegProcess.kill();
             res.destroy();
         });
+
         audio.pipe(ffmpegProcess.stdio[3]).on('error', () => {
             ffmpegProcess.kill();
             res.destroy();
         });
-        
         audio.on('error', () => {
             ffmpegProcess.kill();
             res.destroy();
@@ -77,6 +141,19 @@ export function streamLiveRender(streamInfo, res) {
             res.destroy();
         });
 
+        video.pipe(ffmpegProcess.stdio[5]).on('error', () => {
+            ffmpegProcess.kill();
+            res.destroy();
+        });
+        video.on('error', () => {
+            ffmpegProcess.kill();
+            res.destroy();
+        });
+        video.on('aborted', () => {
+            ffmpegProcess.kill();
+            res.destroy();
+        });
+
         ffmpegProcess.on('disconnect', () => ffmpegProcess.kill());
         ffmpegProcess.on('close', () => ffmpegProcess.kill());
         ffmpegProcess.on('exit', () => ffmpegProcess.kill());
@@ -93,9 +170,10 @@ export function streamLiveRender(streamInfo, res) {
 }
 export function streamAudioOnly(streamInfo, res) {
     try {
+        const usePipedAudio = streamInfo.service === 'youtube';
         let args = [
             '-loglevel', '-8',
-            '-i', streamInfo.urls
+            '-i', usePipedAudio ? 'pipe:4' : streamInfo.urls
         ]
         if (streamInfo.metadata) {
             if (streamInfo.metadata.cover) { // currently corrupts the audio
@@ -117,13 +195,33 @@ export function streamAudioOnly(streamInfo, res) {
             windowsHide: true,
             stdio: [
                 'inherit', 'inherit', 'inherit',
-                'pipe'
+                'pipe', 'pipe'
             ],
         });
         res.setHeader('Connection', 'keep-alive');
         res.setHeader('Content-Disposition', `attachment; filename="${streamInfo.filename}.${streamInfo.audioFormat}"`);
         ffmpegProcess.stdio[3].pipe(res);
 
+        if (usePipedAudio) {
+            const audio = streamInfo.service === 'youtube' ? createYoutubeStream(streamInfo.urls) : got.get(streamInfo.urls, { isStream: true });
+            audio.pipe(ffmpegProcess.stdio[4]).on('error', () => {
+                ffmpegProcess.kill();
+                res.destroy();
+            });
+            audio.on('error', () => {
+                ffmpegProcess.kill();
+                res.destroy();
+            });
+            audio.on('aborted', () => {
+                ffmpegProcess.kill();
+                res.destroy();
+            });
+            res.on('error', () => {
+                ffmpegProcess.kill();
+                res.destroy();
+            });
+        }
+
         ffmpegProcess.on('disconnect', () => ffmpegProcess.kill());
         ffmpegProcess.on('close', () => ffmpegProcess.kill());
         ffmpegProcess.on('exit', () => ffmpegProcess.kill());
@@ -139,9 +237,10 @@ export function streamAudioOnly(streamInfo, res) {
 }
 export function streamVideoOnly(streamInfo, res) {
     try {
+        const usePipedVideo = streamInfo.service === 'youtube';
         let format = streamInfo.filename.split('.')[streamInfo.filename.split('.').length - 1], args = [
             '-loglevel', '-8',
-            '-i', streamInfo.urls,
+            '-i', usePipedVideo ? 'pipe:4' : streamInfo.urls,
             '-c', 'copy'
         ]
         if (streamInfo.mute) args.push('-an');
@@ -152,13 +251,33 @@ export function streamVideoOnly(streamInfo, res) {
             windowsHide: true,
             stdio: [
                 'inherit', 'inherit', 'inherit',
-                'pipe'
+                'pipe', 'pipe'
             ],
         });
         res.setHeader('Connection', 'keep-alive');
         res.setHeader('Content-Disposition', `attachment; filename="${streamInfo.filename.split('.')[0]}${streamInfo.mute ? '_mute' : ''}.${format}"`);
         ffmpegProcess.stdio[3].pipe(res);
 
+        if (usePipedVideo) {
+            const video = streamInfo.service === 'youtube' ? createYoutubeStream(streamInfo.urls) : got.get(streamInfo.urls, { isStream: true });
+            video.pipe(ffmpegProcess.stdio[4]).on('error', () => {
+                ffmpegProcess.kill();
+                res.destroy();
+            });
+            video.on('error', () => {
+                ffmpegProcess.kill();
+                res.destroy();
+            });
+            video.on('aborted', () => {
+                ffmpegProcess.kill();
+                res.destroy();
+            });
+            res.on('error', () => {
+                ffmpegProcess.kill();
+                res.destroy();
+            });
+        }
+
         ffmpegProcess.on('disconnect', () => ffmpegProcess.kill());
         ffmpegProcess.on('close', () => ffmpegProcess.kill());
         ffmpegProcess.on('exit', () => ffmpegProcess.kill());