refactor(download): controller with:

- Event emitters
- Improved error handling.
- Can not force a download from qualities different from current set

server/controller/download.controller.js

@@ -1,15 +1,23 @@
+const EventEmitter = require("events");
 const { fetchDownload, cancelDownload } = require("../services/download.services");
 const { logger } = require("../logger");
 const { isValidUrl, isSafePath } = require("../helpers/validation.helpers");
 const { notifyDownloadFinished } = require("../helpers/notify.helpers");
 const { configFeatures } = require("../../config");
 
+const progressEmitter = new EventEmitter();
+const speedEmitter = new EventEmitter();
+const stageEmitter = new EventEmitter();
+const playlistEmitter = new EventEmitter();
+
+progressEmitter.setMaxListeners(0);
+speedEmitter.setMaxListeners(0);
+stageEmitter.setMaxListeners(0);
+playlistEmitter.setMaxListeners(0);
+
 /**
  * Active Server-Sent Event subscribers receiving
  * download progress updates.
- *
- * Each listener receives a percentage value between
- * 0 and 100 during the download lifecycle.
  */
 const progressListeners = [];
 
@@ -31,6 +39,13 @@ const stageListeners = [];
  */
 const playlistListeners = [];
 
+/**
+ * Set of allowed quality values for downloads.
+ * Used to validate user input and prevent invalid
+ * quality options from being passed to the download service.
+ */
+const ALLOWED_QUALITIES = new Set(["best", "medium", "worst", "1080", "720", "480"]);
+
 /**
  * Handles download requests and starts a yt-dlp download.
  *
@@ -47,34 +62,91 @@ const playlistListeners = [];
  */
 async function downloadController(req, res) {
   try {
+    const body = req.body || {};
     const options = {
-      url: req.body.url,
+      url: body.url,
-      audioOnly: req.body.audioOnly === "1",
+      audioOnly: body.audioOnly === "1" || body.audioOnly === true,
-      quality: req.body.quality || "best",
+      quality: ALLOWED_QUALITIES.has(body.quality) ? body.quality : "best",
-      outputFolder: req.body.savePath || undefined,
+      outputFolder: body.savePath || body.outputFolder || undefined,
-      playlistTitle: req.body.playlistTitle || undefined,
+      playlistTitle: body.playlistTitle || undefined,
     };
 
-    if (!options.url || !isValidUrl(options.url)) return res.status(400).send("Invalid URL !");
+    if (!options.url || typeof options.url !== "string" || !isValidUrl(options.url)) {
+      return res.status(400).json({
+        error: {code: "INVALID_URL", message: "Invalid URL"},
+      });
+    }
     if (options.outputFolder && !isSafePath(options.outputFolder)) {
       logger.warn(`Unsafe download path rejected: ${options.outputFolder}`);
-      return res.status(400).send("Save Path Not Allowed.");
+      return res.status(400).json({
+        error: {code: "INVALID_PATH", message: "Save path not allowed"},
+      });
     }
 
+    const progressHandler = (percent) => {
+      progressEmitter.emit("progress", percent);
+      for (let i = progressListeners.length - 1; i >= 0; i -= 1) {
+        try {
+          progressListeners[i](percent);
+        } catch (err) {
+          logger.debug(`progress listener error: ${err.message}`);
+        }
+      }
+    };
+
+    const speedHandler = (speed) => {
+      speedEmitter.emit("speed", speed);
+      for (let i = speedListeners.length - 1; i >= 0; i -= 1) {
+        try {
+          speedListeners[i](speed);
+        } catch (err) {
+          logger.debug(`speed listener error: ${err.message}`);
+        }
+      }
+    };
+
+    const stageHandler = (stage) => {
+      stageEmitter.emit("stage", stage);
+      for (let i = stageListeners.length - 1; i >= 0; i -= 1) {
+        try {
+          stageListeners[i](stage);
+        } catch (err) {
+          logger.debug(`stage listener error: ${err.message}`);
+        }
+      }
+    };
+
+    const playlistHandler = (info) => {
+      playlistEmitter.emit("playlist", info);
+      for (let i = playlistListeners.length - 1; i >= 0; i -= 1) {
+        try {
+          playlistListeners[i](info);
+        } catch (err) {
+          logger.debug(`playlist listener error: ${err.message}`);
+        }
+      }
+    };
+
     const outputFolder = await fetchDownload(
       options,
-      progressListeners,
+        [progressHandler],
-      speedListeners,
+        [speedHandler],
-      stageListeners,
+        [stageHandler],
-      playlistListeners
+        [playlistHandler]
     );
-    
+
     notifyDownloadFinished(outputFolder, configFeatures.notifySystem);
-    res.send("Download Done !");
+    return res.json({success: true, message: "Download Done !", outputFolder});
-    
   } catch (err) {
     logger.error(`Server Error in /download: ${err?.message ?? err}`);
-    res.status(500).send("Server Error");
+    if (err && err.message === "Download already in progress") {
+      return res.status(409).json({
+        error: {code: "DOWNLOAD_IN_PROGRESS", message: "A download is already in progress"},
+      });
+    }
+    return res.status(500).json({
+      error: {code: "DOWNLOAD_FAILED", message: "Server Error"},
+    });
   }
 }
 
@@ -87,17 +159,33 @@ async function downloadController(req, res) {
  */
 function progressController(req, res) {
   res.set({
-    'Content-Type': 'text/event-stream',
+    "Content-Type": "text/event-stream",
-    'Cache-Control': 'no-cache',
+    "Cache-Control": "no-cache",
-    'Connection': 'keep-alive',
+    Connection: "keep-alive",
+    "X-Accel-Buffering": "no",
   });
+  res.flushHeaders?.();
 
-  const sendProgress = percent => res.write(`data: ${percent}\n\n`);
+  const sendProgress = (percent) => {
+    try {
+      res.write(`data: ${percent}\n\n`);
+    } catch (err) {
+      // ignore broken pipe
+    }
+  };
   progressListeners.push(sendProgress);
+  const onProgress = (percent) => sendProgress(percent);
+  progressEmitter.on("progress", onProgress);
 
-  req.on('close', () => {
+  req.on("close", () => {
-    progressListeners.splice(progressListeners.indexOf(sendProgress), 1);
+    const index = progressListeners.indexOf(sendProgress);
-    res.end();
+    if (index >= 0) progressListeners.splice(index, 1);
+    progressEmitter.removeListener("progress", onProgress);
+    try {
+      res.end();
+    } catch (err) {
+      // ignore
+    }
   });
 }
 
@@ -110,17 +198,33 @@ function progressController(req, res) {
  */
 function speedController(req, res) {
   res.set({
-    'Content-Type': 'text/event-stream',
+    "Content-Type": "text/event-stream",
-    'Cache-Control': 'no-cache',
+    "Cache-Control": "no-cache",
-    'Connection': 'keep-alive',
+    Connection: "keep-alive",
+    "X-Accel-Buffering": "no",
   });
+  res.flushHeaders?.();
 
-  const sendSpeed = speed => res.write(`data: ${speed}\n\n`);
+  const sendSpeed = (speed) => {
+    try {
+      res.write(`data: ${speed}\n\n`);
+    } catch (err) {
+      // ignore broken pipe
+    }
+  };
   speedListeners.push(sendSpeed);
+  const onSpeed = (speed) => sendSpeed(speed);
+  speedEmitter.on("speed", onSpeed);
 
-  req.on('close', () => {
+  req.on("close", () => {
-    speedListeners.splice(speedListeners.indexOf(sendSpeed), 1);
+    const index = speedListeners.indexOf(sendSpeed);
-    res.end();
+    if (index >= 0) speedListeners.splice(index, 1);
+    speedEmitter.removeListener("speed", onSpeed);
+    try {
+      res.end();
+    } catch (err) {
+      // ignore
+    }
   });
 }
 
@@ -134,17 +238,33 @@ function speedController(req, res) {
  */
 function stageController(req, res) {
   res.set({
-    'Content-Type': 'text/event-stream',
+    "Content-Type": "text/event-stream",
-    'Cache-Control': 'no-cache',
+    "Cache-Control": "no-cache",
-    'Connection': 'keep-alive',
+    Connection: "keep-alive",
+    "X-Accel-Buffering": "no",
   });
+  res.flushHeaders?.();
 
-  const sendStage = stage => res.write(`data: ${stage}\n\n`);
+  const sendStage = (stage) => {
+    try {
+      res.write(`data: ${stage}\n\n`);
+    } catch (err) {
+      // ignore broken pipe
+    }
+  };
   stageListeners.push(sendStage);
+  const onStage = (stage) => sendStage(stage);
+  stageEmitter.on("stage", onStage);
 
-  req.on('close', () => {
+  req.on("close", () => {
-    stageListeners.splice(stageListeners.indexOf(sendStage), 1);
+    const index = stageListeners.indexOf(sendStage);
-    res.end();
+    if (index >= 0) stageListeners.splice(index, 1);
+    stageEmitter.removeListener("stage", onStage);
+    try {
+      res.end();
+    } catch (err) {
+      // ignore
+    }
   });
 }
 
@@ -160,10 +280,12 @@ function cancelDownloadController(req, res) {
   const cancelled = cancelDownload();
   if (cancelled) {
     logger.info("Download and queue cancelled by user");
-    res.send("Download stopped! Queue cleared.");
+    return res.json({success: true, message: "Download stopped! Queue cleared."});
   } else {
     logger.warn("No download to cancel");
-    res.status(400).send("No download in progress !");
+    return res.status(400).json({
+      error: {code: "NO_DOWNLOAD_IN_PROGRESS", message: "No download in progress !"},
+    });
   }
 }
 
@@ -181,18 +303,42 @@ function cancelDownloadController(req, res) {
  */
 function playlistInfoController(req, res) {
   res.set({
-    'Content-Type': 'text/event-stream',
+    "Content-Type": "text/event-stream",
-    'Cache-Control': 'no-cache',
+    "Cache-Control": "no-cache",
-    'Connection': 'keep-alive',
+    Connection: "keep-alive",
+    "X-Accel-Buffering": "no",
   });
+  res.flushHeaders?.();
 
-  const sendPlaylistInfo = info => res.write(`data: ${info}\n\n`);
+  const sendPlaylistInfo = (info) => {
+    try {
+      res.write(`data: ${info}\n\n`);
+    } catch (err) {
+      // ignore broken pipe
+    }
+  };
   playlistListeners.push(sendPlaylistInfo);
+  const onPlaylist = (info) => sendPlaylistInfo(info);
+  playlistEmitter.on("playlist", onPlaylist);
 
-  req.on('close', () => {
+  // Remove listener when client disconnects
-    playlistListeners.splice(playlistListeners.indexOf(sendPlaylistInfo), 1);
+  req.on("close", () => {
-    res.end();
+    const index = playlistListeners.indexOf(sendPlaylistInfo);
+    if (index >= 0) playlistListeners.splice(index, 1);
+    playlistEmitter.removeListener("playlist", onPlaylist);
+    try {
+      res.end();
+    } catch (err) {
+      // ignore
+    }
   });
 }
 
-module.exports = { downloadController, progressController, speedController, stageController, cancelDownloadController, playlistInfoController};
+module.exports = {
+  downloadController,
+  progressController,
+  speedController,
+  stageController,
+  cancelDownloadController,
+  playlistInfoController,
+};