302 lines
9.5 KiB
JavaScript
302 lines
9.5 KiB
JavaScript
'use strict';
|
|
|
|
const path = require('path');
|
|
const fs = require('fs');
|
|
|
|
/* eslint-disable no-param-reassign */
|
|
|
|
const MAX_RETRIES = 2;
|
|
const RETRY_DELAY_MS = 1000;
|
|
const PROGRESS_INTERVAL_MS = 1000;
|
|
|
|
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. Self-contained to avoid external helper dependencies.
|
|
*/
|
|
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 lastProgressTime = 0;
|
|
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);
|
|
|
|
// Update UI and jobLog every second (PROGRESS_INTERVAL_MS) or on initial data chunk
|
|
const timeSinceLastProgress = now - lastProgressTime;
|
|
if (lastProgressTime === 0 || timeSinceLastProgress >= PROGRESS_INTERVAL_MS) {
|
|
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);
|
|
|
|
if (typeof updateWorker === 'function') {
|
|
updateWorker({
|
|
percentage: percent,
|
|
ETA: eta,
|
|
});
|
|
}
|
|
if (typeof updateProgress === 'function') {
|
|
updateProgress({ progress: percent });
|
|
}
|
|
|
|
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`,
|
|
);
|
|
}
|
|
|
|
lastProgressTime = 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}`,
|
|
);
|
|
};
|
|
|
|
/**
|
|
* copyToWorkDir
|
|
* Copies the source file into the worker's local workDir before processing.
|
|
*
|
|
* Use this plugin BEFORE any transcode/extract plugin when:
|
|
* - The source file lives on a network share / mapped drive
|
|
* - You want the worker to encode from a local copy to avoid I/O bottleneck
|
|
* - The network path is slow or unreliable during long encodes
|
|
*
|
|
* Flow:
|
|
* 1. Copies source -> workDir/<basename>_old<ext> (streamed with real-time progress)
|
|
* 2. Updates outputFileObj._id to the local copy path
|
|
* 3. Downstream plugins operate on the local copy
|
|
* 4. Tdarr replaces the original with the final output via replaceOriginalFile
|
|
*/
|
|
|
|
const details = () => ({
|
|
name: 'Copy Source to Work Dir',
|
|
description:
|
|
'Copies the source file into the worker local workDir using a streamed, verified copy '
|
|
+ 'with real-time progress, speed, and ETA tracking. Use before transcode plugins '
|
|
+ 'when the source is on a network share to avoid I/O bottlenecks during encoding.',
|
|
style: { borderColor: '#607D8B' },
|
|
tags: 'utility,copy,progress',
|
|
isStartPlugin: false,
|
|
pType: '',
|
|
requiresVersion: '2.11.01',
|
|
sidebarPosition: 0,
|
|
icon: 'faCopy',
|
|
inputs: [
|
|
{
|
|
label: 'Skip if already local',
|
|
name: 'skipIfLocal',
|
|
type: 'boolean',
|
|
defaultValue: 'true',
|
|
inputUI: {
|
|
type: 'dropdown',
|
|
options: ['true', 'false'],
|
|
},
|
|
tooltip:
|
|
'If true, skip the copy when the source file is already inside workDir '
|
|
+ 'or on the same drive as workDir (avoids double-copy on mapped nodes).',
|
|
},
|
|
],
|
|
outputs: [
|
|
{ number: 1, tooltip: 'File is ready in workDir — continue to next plugin' },
|
|
],
|
|
});
|
|
|
|
const plugin = async (args) => {
|
|
const { jobLog, inputFileObj, variables, workDir, updateWorker } = args;
|
|
|
|
const inputFile = inputFileObj._id;
|
|
const ext = path.extname(inputFile);
|
|
const baseName = path.basename(inputFile, ext);
|
|
const localCopy = path.join(workDir, `${baseName}_old${ext}`);
|
|
const skipIfLocal = (args.inputs.skipIfLocal || 'true') === 'true';
|
|
|
|
jobLog(`[copyToWorkDir] Source : ${inputFile}\n`);
|
|
jobLog(`[copyToWorkDir] WorkDir: ${workDir}\n`);
|
|
|
|
// Skip if source file is already inside workDir
|
|
const relativeInput = path.relative(path.resolve(workDir), path.resolve(inputFile));
|
|
if (
|
|
skipIfLocal
|
|
&& relativeInput !== '..'
|
|
&& !relativeInput.startsWith(`..${path.sep}`)
|
|
&& !path.isAbsolute(relativeInput)
|
|
) {
|
|
jobLog('[copyToWorkDir] Source file is already inside workDir — skipping copy.\n');
|
|
return { outputNumber: 1, outputFileObj: inputFileObj, variables };
|
|
}
|
|
|
|
if (!fs.existsSync(inputFile)) {
|
|
throw new Error(`[copyToWorkDir] Source file not found: ${inputFile}`);
|
|
}
|
|
|
|
// Preserve original library paths for downstream plugins
|
|
if (!variables.originalLibraryFile) {
|
|
variables.originalLibraryFile = inputFile;
|
|
variables.originalLibraryDir = path.dirname(inputFile);
|
|
}
|
|
variables.workDirSourceBaseName = baseName;
|
|
variables.workDirCopyApplied = 'true';
|
|
|
|
jobLog(`[copyToWorkDir] Copying ${inputFile} to ${localCopy}\n`);
|
|
await copyWithProgress({
|
|
source: inputFile,
|
|
destination: localCopy,
|
|
jobLog,
|
|
updateWorker,
|
|
updateProgress: args.updateProgress,
|
|
label: 'copyToWorkDir',
|
|
});
|
|
|
|
return {
|
|
outputNumber: 1,
|
|
outputFileObj: { ...inputFileObj, _id: localCopy },
|
|
variables,
|
|
};
|
|
};
|
|
|
|
module.exports.details = details;
|
|
module.exports.plugin = plugin;
|
|
module.exports.copyWithProgress = copyWithProgress;
|