Tdarr-plugin/FlowHelpers/1.0.0/copyWithProgress.js

'use strict';

const fs = require('fs');

const copyWithProgress = async ({
  source,
  destination,
  jobLog,
  updateProgress,
  label,
  copyMethod = 'auto',
  bufferSizeMb = 16,
  maxRetries = 2,
  retryDelayMs = 1000,
}) => {
  const sourceSize = fs.statSync(source).size;
  const sourceSizeMb = (sourceSize / 1024 / 1024).toFixed(1);
  const normalizedBufferSizeMb = Math.min(
    Math.max(Number.parseInt(bufferSizeMb, 10) || 16, 1),
    128,
  );
  const bufferSize = normalizedBufferSizeMb * 1024 * 1024;

  const report = (percent, message) => {
    const boundedPercent = Math.min(Math.max(percent, 0), 100);
    if (typeof updateProgress === 'function') {
      updateProgress({
        progress: boundedPercent,
        jobLog: message,
      });
    }
    if (typeof jobLog === 'function') jobLog(`${message}\n`);
  };

  const copyOnceWithStream = () => new Promise((resolve, reject) => {
    const startMs = Date.now();
    let copiedBytes = 0;
    let nextProgressPercent = 0;
    const rd = fs.createReadStream(source, { highWaterMark: bufferSize });
    const wr = fs.createWriteStream(destination, { highWaterMark: bufferSize });
    let settled = false;
    const fail = (error) => {
      if (settled) return;
      settled = true;
      rd.destroy();
      wr.destroy();
      reject(error);
    };
    rd.on('error', fail);
    wr.on('error', fail);
    wr.on('finish', () => {
      if (settled) return;
      settled = true;
      resolve(((Date.now() - startMs) / 1000).toFixed(1));
    });
    rd.on('data', (chunk) => {
      copiedBytes += chunk.length;
      const percent = sourceSize === 0 ? 100 : Math.floor((copiedBytes / sourceSize) * 100);
      if (percent < nextProgressPercent) return;
      const elapsedSeconds = Math.max((Date.now() - startMs) / 1000, 0.1);
      report(
        percent,
        `[${label}] Progress: ${Math.min(percent, 100)}% `
          + `(${(copiedBytes / 1024 / 1024).toFixed(1)}/${sourceSizeMb} MB, `
          + `${(copiedBytes / 1024 / 1024 / elapsedSeconds).toFixed(1)} MB/s)`,
      );
      nextProgressPercent = (Math.floor(percent / 10) + 1) * 10;
    });
    rd.pipe(wr);
  });

  const copyOnceWithNative = () => new Promise((resolve, reject) => {
    const startMs = Date.now();
    let settled = false;
    const progressTimer = setInterval(() => {
      if (settled || !fs.existsSync(destination)) return;
      const copiedBytes = fs.statSync(destination).size;
      const elapsedSeconds = Math.max((Date.now() - startMs) / 1000, 0.1);
      report(
        Math.floor((copiedBytes / sourceSize) * 100),
        `[${label}] Progress: ${Math.min(Math.floor((copiedBytes / sourceSize) * 100), 100)}% `
          + `(${(copiedBytes / 1024 / 1024).toFixed(1)}/${sourceSizeMb} MB, `
          + `${(copiedBytes / 1024 / 1024 / elapsedSeconds).toFixed(1)} MB/s)`,
      );
    }, 1000);
    fs.promises.copyFile(source, destination).then(() => {
      settled = true;
      clearInterval(progressTimer);
      resolve(((Date.now() - startMs) / 1000).toFixed(1));
    }).catch((error) => {
      settled = true;
      clearInterval(progressTimer);
      reject(error);
    });
  });

  const copyOnce = copyMethod === 'stream' ? copyOnceWithStream : copyOnceWithNative;

  let lastError;
  for (let attempt = 0; attempt <= maxRetries; attempt += 1) {
    if (attempt > 0) {
      report(0, `[${label}] Retrying copy (${attempt}/${maxRetries})...`);
      await new Promise((resolve) => setTimeout(resolve, retryDelayMs));
    }
    try {
      if (fs.existsSync(destination)) fs.unlinkSync(destination);
      const elapsed = await copyOnce();
      const destinationSize = fs.statSync(destination).size;
      if (destinationSize !== sourceSize) {
        throw new Error(
          `Copy size mismatch: expected ${sourceSize} bytes, got ${destinationSize}`,
        );
      }
      report(100, `[${label}] Done: ${sourceSizeMb} MB copied in ${elapsed}s`);
      return;
    } catch (error) {
      lastError = error;
      try {
        if (fs.existsSync(destination)) fs.unlinkSync(destination);
      } catch (cleanupError) {
        lastError = new Error(
          `${error.message}; could not remove partial copy: ${cleanupError.message}`,
        );
      }
      if (attempt === maxRetries) break;
      if (typeof jobLog === 'function') {
        jobLog(`[${label}] Copy attempt ${attempt + 1} failed: ${error.message}\n`);
      }
    }
  }

  throw new Error(
    `[${label}] Copy failed after ${maxRetries + 1} attempts: ${lastError.message}`,
  );
};

module.exports = { copyWithProgress };