repiped/routes/prepare.js

const fs = require("fs");
const path = require("path");
const { exec } = require("child_process");
const isWindows = process.platform === "win32";
const isArm = process.arch === "arm" || process.arch === "arm64";
const isAarch64 = process.arch === "arm64";

const ytDlpPath = path.join(
  "deps",
  isWindows ? "yt-dlp.exe" : 
  isArm ? (isAarch64 ? "yt-dlp_linux_aarch64" : "yt-dlp_linux_armv7l") : 
  "yt-dlp"
);

if (!isWindows) {
  fs.chmodSync(ytDlpPath, 0o755);
}

process.env.PATH = `${path.join(process.cwd(), "deps")}${path.delimiter}${process.env.PATH}`;


const activeVideoProcesses = new Map();
const PROCESS_TIMEOUT = 30 * 60 * 1000;

function logWithVideoId(videoId, message, isError = false) {
  const timestamp = new Date().toISOString();
  const logFn = isError ? console.error : console.log;
  logFn(`[${timestamp}][${videoId || "NO_VIDEO_ID"}] ${message}`);
}

async function processVideo(videoId) {
  function updateProgress(percentage, state) {
    if (activeVideoProcesses.has(videoId)) {
      const processInfo = activeVideoProcesses.get(videoId);
      processInfo.progress = percentage;
      if (state) {
        processInfo.state = state;
      }

      logWithVideoId(
        videoId,
        `Progress update: ${percentage.toFixed(2)}% (${
          state || processInfo.state
        })`
      );

      for (const client of processInfo.clients) {
        if (client.ws.readyState === 1) {
          client.ws.send(JSON.stringify({ preparing: percentage }));
        }
      }
    }
  }

  logWithVideoId(videoId, "Starting video download");
  updateProgress(0, "downloading");

  return new Promise((resolve, reject) => {
    try {
      const videoDirectory = `videohost/${videoId}`;
      if (!fs.existsSync(videoDirectory)) {
        fs.mkdirSync(videoDirectory, { recursive: true });
      }
      const process = exec(
        `${ytDlpPath} --cookies cookies.txt --force-ipv4 --prefer-insecure -f "bestvideo[ext=mp4][height<=1440][vcodec^=avc]+bestaudio[ext=m4a]" --merge-output-format mp4 -o "videohost/${videoId}.mp4" "https://youtube.com/watch?v=${videoId}"`
      );

      console.log(
        `${ytDlpPath} --cookies cookies.txt --force-ipv4 --prefer-insecure -f "bestvideo[ext=mp4][height<=1440][vcodec^=avc]+bestaudio[ext=m4a]" --merge-output-format mp4 -o "videohost/${videoId}.mp4" "https://youtube.com/watch?v=${videoId}"`
      );

      let isFirstDownload = true;
      let currentDestination = "";

      process.stdout.on("data", (data) => {
        const output = data.toString();

        if (output.includes("[download] Destination:")) {
          isFirstDownload = !currentDestination;
          currentDestination = output;
        }

        const match = output.match(/\[download\]\s+(\d+\.\d+)% of/);
        if (match) {
          let percentage = parseFloat(match[1]);
          if (!isFirstDownload) {
            percentage = 72 + percentage * 0.08;
          } else {
            percentage = percentage * 0.72;
          }
          updateProgress(percentage, "downloading");
        }
      });

      process.stderr.on("data", (data) => {
        logWithVideoId(videoId, `yt-dlp stderr: ${data}`, true);
      });

      process.on("close", async (code) => {
        if (code === 0) {
          logWithVideoId(
            videoId,
            "Download completed, starting conversion to HLS"
          );
          updateProgress(80, "preparing");

          await new Promise(resolve => setTimeout(resolve, 1000));

          logWithVideoId(videoId, "Starting conversion to HLS");
          updateProgress(80, "converting");

          try {
            await convertToHLS(videoId, updateProgress);

            fs.unlink(`videohost/${videoId}.mp4`, (err) => {
              if (err) {
                logWithVideoId(
                  videoId,
                  `Error deleting original file: ${err.message}`,
                  true
                );
              } else {
                logWithVideoId(
                  videoId,
                  "Original mp4 file deleted successfully"
                );
              }
            });

            resolve();
          } catch (error) {
            reject(error);
          }
        } else {
          logWithVideoId(
            videoId,
            `Download process exited with code ${code}`,
            true
          );
          reject(new Error("Unable to prepare video. Please try again later."));
        }
      });
    } catch (error) {
      reject(new Error(`Failed to start download process: ${error.message}`));
    }
  });
}

async function convertToHLS(videoId, updateProgress) {
  logWithVideoId(videoId, "Starting HLS conversion");

  return new Promise((resolve, reject) => {
    try {
      const videoDirectory = `videohost/${videoId}`;

      const conversionProcess = exec(
        `ffmpeg -i "videohost/${videoId}.mp4" -c:v copy -c:a copy -hls_time 6 -hls_list_size 0 -hls_segment_type mpegts -hls_playlist_type vod -hls_segment_filename "${videoDirectory}/%03d.ts" "${videoDirectory}/playlist.m3u8"`
      );

      let totalDuration = 0;

      conversionProcess.stderr.on("data", (data) => {
        const output = data.toString();
        console.log("ffmpeg stderr:", output);

        const durationMatch = output.match(
          /Duration: (\d{2}):(\d{2}):(\d{2})\.(\d{2})/
        );
        if (durationMatch) {
          const [_, hours, minutes, seconds, centiseconds] = durationMatch;
          totalDuration =
            parseFloat(hours) * 3600 +
            parseFloat(minutes) * 60 +
            parseFloat(seconds) +
            parseFloat(centiseconds) / 100;

          logWithVideoId(
            videoId,
            `Video duration: ${totalDuration.toFixed(2)} seconds`
          );
        }

        const timeMatch = output.match(/time=(\d{2}):(\d{2}):(\d{2})\.(\d{2})/);
        if (timeMatch && totalDuration > 0) {
          const [_, hours, minutes, seconds, centiseconds] = timeMatch;
          const currentTime =
            parseFloat(hours) * 3600 +
            parseFloat(minutes) * 60 +
            parseFloat(seconds) +
            parseFloat(centiseconds) / 100;

          const percentage = 80 + (currentTime / totalDuration) * 20;
          updateProgress(Math.min(percentage, 100), "converting");
        }
      });

      conversionProcess.stdout.on("data", (data) => {
        logWithVideoId(videoId, `ffmpeg stdout: ${data}`);
      });

      conversionProcess.on("close", (code) => {
        if (code === 0) {
          logWithVideoId(videoId, "Conversion completed successfully");
          updateProgress(100, "completed");
          resolve();
        } else {
          reject(new Error(`Conversion process exited with code ${code}`));
        }
      });
    } catch (error) {
      reject(new Error(`Failed to start conversion process: ${error.message}`));
    }
  });
}


function setupCleanupInterval() {
  setInterval(() => {
    const now = Date.now();
    for (const [videoId, info] of activeVideoProcesses.entries()) {
      if (now - info.startTime > PROCESS_TIMEOUT) {
        logWithVideoId(
          videoId,
          "Cleaning up stalled process in periodic check",
          true
        );

        for (const client of info.clients) {
          if (client.ws.readyState === 1) {
            client.ws.send(
              JSON.stringify({
                ok: false,
                error: "Video processing timed out",
              })
            );
          }
        }

        if (info.timeoutId) {
          clearTimeout(info.timeoutId);
        }

        activeVideoProcesses.delete(videoId);
      }
    }
  }, 5 * 60 * 1000);
}


function prepareRouteHandler(app) {
  app.ws("/api/v1/prepare", (ws, req) => {
    let videoId = null;
    let clientId = Date.now() + Math.random().toString(36).substring(2, 15);

    logWithVideoId(null, `New WebSocket connection established (${clientId})`);

    ws.on("message", async (msg) => {
      try {
        const parsedMessage = JSON.parse(msg);

        if (parsedMessage.videoId) {
          videoId = parsedMessage.videoId;
          ws.send(JSON.stringify({ ok: true }));
          logWithVideoId(videoId, `Request for video (client: ${clientId})`);

          const isValidYoutubeId = /^[a-zA-Z0-9_-]{11}$/.test(videoId);
          if (!isValidYoutubeId) {
            ws.send(
              JSON.stringify({ ok: false, error: "Invalid YouTube video ID" })
            );
            logWithVideoId(videoId, "Invalid YouTube video ID", true);
            return;
          }

          const videoPath = `videohost/${videoId}/playlist.m3u8`;
          if (fs.existsSync(videoPath)) {
            logWithVideoId(videoId, "Video already exists - sending stream URL");
            ws.send(
              JSON.stringify({
                done: true,
                streamUrl: "/videocd/" + videoId + "/playlist.m3u8",
              })
            );
            return;
          }

          if (activeVideoProcesses.has(videoId)) {
            const processInfo = activeVideoProcesses.get(videoId);
            logWithVideoId(
              videoId,
              `Video is already being processed (state: ${
                processInfo.state
              }, progress: ${processInfo.progress.toFixed(
                2
              )}%), attaching client ${clientId}`
            );

            processInfo.clients.push({ ws, clientId });

            ws.send(JSON.stringify({ preparing: processInfo.progress }));
            return;
          }

          const processInfo = {
            clients: [{ ws, clientId }],
            state: "initializing",
            progress: 0,
            startTime: Date.now(),
            timeoutId: null,
          };

          processInfo.timeoutId = setTimeout(() => {
            if (activeVideoProcesses.has(videoId)) {
              logWithVideoId(videoId, "Process timed out after 30 minutes", true);

              const info = activeVideoProcesses.get(videoId);
              for (const client of info.clients) {
                if (client.ws.readyState === 1) {
                  client.ws.send(
                    JSON.stringify({
                      ok: false,
                      error: "Video processing timed out",
                    })
                  );
                }
              }

              activeVideoProcesses.delete(videoId);
            }
          }, PROCESS_TIMEOUT);

          activeVideoProcesses.set(videoId, processInfo);
          logWithVideoId(videoId, "Starting new video processing");

          try {
            await processVideo(videoId);

            if (processInfo.timeoutId) {
              clearTimeout(processInfo.timeoutId);
            }

            if (activeVideoProcesses.has(videoId)) {
              const info = activeVideoProcesses.get(videoId);
              logWithVideoId(
                videoId,
                `Processing completed. Notifying ${info.clients.length} clients`
              );

              for (const client of info.clients) {
                if (client.ws.readyState === 1) {
                  client.ws.send(
                    JSON.stringify({
                      done: true,
                      streamUrl: "/videocd/" + videoId + "/playlist.m3u8",
                    })
                  );
                }
              }

              activeVideoProcesses.delete(videoId);
            }

            logWithVideoId(videoId, "Video prepared successfully");
          } catch (error) {
            if (processInfo.timeoutId) {
              clearTimeout(processInfo.timeoutId);
            }

            logWithVideoId(
              videoId,
              `Error processing video: ${error.message}`,
              true
            );

            if (activeVideoProcesses.has(videoId)) {
              const info = activeVideoProcesses.get(videoId);
              for (const client of info.clients) {
                if (client.ws.readyState === 1) {
                  client.ws.send(
                    JSON.stringify({
                      ok: false,
                      error:
                        "Unable to prepare video. Please try reloading the page.",
                    })
                  );
                }
              }

              activeVideoProcesses.delete(videoId);
            }
          }
        }
      } catch (error) {
        logWithVideoId(videoId, `Unexpected error: ${error.message}`, true);
        ws.send(JSON.stringify({ ok: false, error: "Server error occurred" }));
      }
    });

    ws.on("close", () => {
      logWithVideoId(videoId, `WebSocket closed for client ${clientId}`);

      if (videoId && activeVideoProcesses.has(videoId)) {
        const processInfo = activeVideoProcesses.get(videoId);
        const clientIndex = processInfo.clients.findIndex(
          (client) => client.clientId === clientId
        );

        if (clientIndex !== -1) {
          processInfo.clients.splice(clientIndex, 1);
          logWithVideoId(
            videoId,
            `Removed client ${clientId} from process. Remaining clients: ${processInfo.clients.length}`
          );
        }
      }
    });

    ws.on("error", (error) => {
      logWithVideoId(
        videoId,
        `WebSocket error for client ${clientId}: ${error.message}`,
        true
      );
    });
  });
  
  
  setupCleanupInterval();
}

module.exports = prepareRouteHandler;