Tdarr-plugin/FlowHelpers/1.0.0/copyWithProgress.js

'use strict';

const fs   = require('fs');
const path = require('path');

const MAX_RETRIES = 2;
const RETRY_DELAY_MS = 1000;
const LOG_INTERVAL_MS = 3000;

const sleep = (milliseconds) => new Promise((resolve) => {
  setTimeout(resolve, milliseconds);
});

const formatEta = (seconds) => {
  if (!Number.isFinite(seconds) || seconds < 0) return '00:00:00';
  const hrs = Math.floor(seconds / 3600);
  const mins = Math.floor((seconds % 3600) / 60);
  const secs = Math.floor(seconds % 60);
  return `${String(hrs).padStart(2, '0')}:${String(mins).padStart(2, '0')}:${String(secs).padStart(2, '0')}`;
};

/**
 * Copies a file from source to destination with real-time stream-based
 * progress reporting, speed / ETA tracking, Tdarr worker updates, size verification,
 * and automatic retry with cleanup.
 */
const copyWithProgress = async ({
  source,
  destination,
  jobLog,
  updateWorker,
  updateProgress,
  label = 'copy',
}) => {
  if (!fs.existsSync(source)) {
    throw new Error(`[${label}] Source file not found: ${source}`);
  }

  const sourceStat = fs.statSync(source);
  const sourceSize = sourceStat.size;
  const sourceSizeMb = sourceSize / 1024 / 1024;

  const destDir = path.dirname(destination);
  if (!fs.existsSync(destDir)) {
    fs.mkdirSync(destDir, { recursive: true });
  }

  const report = (percent, message, eta) => {
    if (typeof updateWorker === 'function') {
      updateWorker({
        percentage: Math.min(percent, 100),
        ETA: eta || '00:00:00',
      });
    }
    if (typeof updateProgress === 'function') {
      updateProgress({
        progress: Math.min(percent, 100),
        jobLog: message,
      });
    }
    if (message && typeof jobLog === 'function') {
      jobLog(`${message}\n`);
    }
  };

  let lastError;

  for (let attempt = 0; attempt <= MAX_RETRIES; attempt += 1) {
    if (attempt > 0) {
      report(0, `[${label}] Retrying copy (${attempt}/${MAX_RETRIES})...`);
      await sleep(RETRY_DELAY_MS);
    }

    try {
      if (fs.existsSync(destination)) {
        fs.unlinkSync(destination);
      }

      // Special case: empty file (0 bytes)
      if (sourceSize === 0) {
        fs.writeFileSync(destination, Buffer.alloc(0));
        report(100, `[${label}] Copied 0-byte file successfully.`, '00:00:00');
        return;
      }

      const startedAt = Date.now();
      let copiedBytes = 0;
      let lastReportedPercent = -1;
      let lastLoggedPercent = -10;
      let lastLoggedTime = startedAt;
      let intervalBytes = 0;
      let intervalTime = startedAt;

      await new Promise((resolve, reject) => {
        const rd = fs.createReadStream(source, { highWaterMark: 1024 * 1024 });
        const wr = fs.createWriteStream(destination, { highWaterMark: 1024 * 1024 });

        rd.on('error', (err) => {
          wr.destroy();
          reject(err);
        });

        wr.on('error', (err) => {
          rd.destroy();
          reject(err);
        });

        wr.on('finish', resolve);

        rd.on('data', (chunk) => {
          copiedBytes += chunk.length;
          intervalBytes += chunk.length;

          const now = Date.now();
          const percent = Math.min(Math.floor((copiedBytes / sourceSize) * 100), 99);

          // Calculate ETA and speed
          const totalElapsedSec = Math.max((now - startedAt) / 1000, 0.001);
          const avgSpeedBps = copiedBytes / totalElapsedSec;
          const remainingBytes = Math.max(0, sourceSize - copiedBytes);
          const remainingSec = avgSpeedBps > 0 ? remainingBytes / avgSpeedBps : 0;
          const eta = formatEta(remainingSec);

          // Update Tdarr UI on every percentage increase
          if (percent > lastReportedPercent) {
            if (typeof updateWorker === 'function') {
              updateWorker({
                percentage: percent,
                ETA: eta,
              });
            }
            if (typeof updateProgress === 'function') {
              updateProgress({ progress: percent });
            }
            lastReportedPercent = percent;
          }

          // Throttle jobLog output: every 10% or every LOG_INTERVAL_MS
          const timeSinceLastLog = now - lastLoggedTime;
          if (percent >= lastLoggedPercent + 10 || (timeSinceLastLog >= LOG_INTERVAL_MS && percent !== lastLoggedPercent)) {
            const intSec = Math.max((now - intervalTime) / 1000, 0.001);
            const currentSpeedMb = (intervalBytes / 1024 / 1024 / intSec).toFixed(1);
            const copiedMb = (copiedBytes / 1024 / 1024).toFixed(1);

            if (typeof jobLog === 'function') {
              jobLog(
                `[${label}] Progress: ${percent}% `
                + `(${copiedMb}/${sourceSizeMb.toFixed(1)} MB, `
                + `${currentSpeedMb} MB/s, ETA: ${eta})\n`,
              );
            }

            lastLoggedPercent = percent;
            lastLoggedTime = now;
            intervalBytes = 0;
            intervalTime = now;
          }
        });

        rd.pipe(wr);
      });

      // Verify destination size
      if (!fs.existsSync(destination)) {
        throw new Error(`Destination file does not exist after copy: ${destination}`);
      }
      const destinationSize = fs.statSync(destination).size;
      if (destinationSize !== sourceSize) {
        throw new Error(
          `Copy size mismatch: expected ${sourceSize} bytes, got ${destinationSize}`,
        );
      }

      const totalElapsed = Math.max((Date.now() - startedAt) / 1000, 0.001);
      const avgSpeed = (sourceSizeMb / totalElapsed).toFixed(1);
      report(
        100,
        `[${label}] Done: ${sourceSizeMb.toFixed(1)} MB copied in `
          + `${totalElapsed.toFixed(1)}s (${avgSpeed} MB/s average)`,
        '00:00:00',
      );
      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 < MAX_RETRIES && typeof jobLog === 'function') {
        jobLog(`[${label}] Copy attempt ${attempt + 1} failed: ${lastError.message}\n`);
      }
    }
  }

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

module.exports = { copyWithProgress };