Files
webshare-api/src/qbt.js
Michal Pemcak 5b71d9f54e fix: TWO_HOURS constant in staging cleanup was actually 5 minutes
Cleanup deleted completed staging files as soon as 5 minutes had
passed instead of the intended 2 hours, so successfully downloaded
files could be silently removed before being moved out of staging.
2026-08-29 04:17:08 +02:00

615 lines
25 KiB
JavaScript

const https = require('https');
const http = require('http');
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const multer = require('multer');
const { extractWebseed, extractInfoHash } = require('./torrent');
const MEDIA_ROOT = process.env.MEDIA_ROOT || '/data';
// Single staging dir for UI + *arr. Sonarr/Radarr import into Shows/Movies.
const SAVE_PATH = process.env.DOWNLOAD_PATH || path.join(MEDIA_ROOT, 'webshare');
const FILE_UID = parseInt(process.env.DOWNLOAD_UID || '1000', 10);
const FILE_GID = parseInt(process.env.DOWNLOAD_GID || '1000', 10);
const upload = multer({ storage: multer.memoryStorage() });
function safeFileName(name) {
return path.basename(String(name || 'download.bin'))
.replace(/[\/\\?%*:|"<>]/g, '_')
.replace(/\s+/g, ' ')
.trim() || 'download.bin';
}
function entryDir(entry) {
return entry.saveDir || SAVE_PATH;
}
function entryPath(entry) {
return path.join(entryDir(entry), entry.name);
}
function ensureDir(dir) {
fs.mkdirSync(dir, { recursive: true });
try {
if (typeof process.getuid === 'function' && process.getuid() === 0
&& Number.isFinite(FILE_UID) && Number.isFinite(FILE_GID)) {
fs.chownSync(dir, FILE_UID, FILE_GID);
}
} catch { /* ignore */ }
}
function fixOwnership(filePath) {
try {
if (typeof process.getuid === 'function' && process.getuid() === 0
&& Number.isFinite(FILE_UID) && Number.isFinite(FILE_GID)
&& fs.existsSync(filePath)) {
fs.chownSync(filePath, FILE_UID, FILE_GID);
try { fs.chownSync(path.dirname(filePath), FILE_UID, FILE_GID); } catch { /* ignore */ }
}
} catch { /* ignore */ }
}
function formatSpeed(bps) {
if (bps >= 1e9) return (bps / 1e9).toFixed(1) + ' GB/s';
if (bps >= 1e6) return (bps / 1e6).toFixed(1) + ' MB/s';
if (bps >= 1e3) return (bps / 1e3).toFixed(0) + ' KB/s';
return bps.toFixed(0) + ' B/s';
}
const MAX_CONCURRENT = parseInt(process.env.MAX_CONCURRENT_DOWNLOADS || '2', 10);
const STATE_FILE = path.join(SAVE_PATH, '.queue-state.json');
// hash → download entry (persisted to disk)
const downloads = new Map(); // hash → entry
const activeRequests = new Map(); // hash → request (destroy to abort)
const cancelled = new Set(); // hashes cancelled while in-flight (skip retry / error bookkeeping)
let activeDownloads = 0;
const waitingQueue = []; // { entry, webseedUrl }
function isCancelled(hash) {
return cancelled.has(hash);
}
function saveState() {
try {
ensureDir(SAVE_PATH);
const data = [...downloads.values()].map(e => ({
hash: e.hash, name: e.name, size: e.size,
progress: e.progress, state: e.state, category: e.category,
webseedUrl: e.webseedUrl,
saveDir: e.saveDir,
}));
fs.writeFileSync(STATE_FILE, JSON.stringify(data));
} catch { }
}
function loadState() {
try {
const data = JSON.parse(fs.readFileSync(STATE_FILE, 'utf8'));
for (const e of data) {
if (e.state === 'downloading' || e.state === 'queuedDL') {
// Always re-download — partial files from interrupted downloads are corrupt
const filePath = path.join(SAVE_PATH, e.name);
try { fs.unlinkSync(filePath); } catch { }
if (e.webseedUrl) {
e.progress = 0;
downloads.set(e.hash, e);
enqueue(e, e.webseedUrl);
console.log(`[qbt] re-queued: ${e.name}`);
}
} else if (e.state === 'uploading') {
const filePath = path.join(SAVE_PATH, e.name);
if (fs.existsSync(filePath)) {
// Sync size to actual file — Content-Length may have been wrong
try { e.size = fs.statSync(filePath).size; } catch {}
downloads.set(e.hash, e);
console.log(`[qbt] restored uploading: ${e.name}`);
}
// File gone → already imported and cleaned up; drop entry.
// Sonarr tracks hasFile=true so it won't re-search.
} else if (e.state === 'error') {
// Keep as error — don't auto-retry on restart; Sonarr will re-search if needed.
downloads.set(e.hash, e);
setTimeout(() => {
downloads.delete(e.hash);
saveState();
}, 3 * 60 * 1000);
} else {
downloads.set(e.hash, e);
}
}
} catch { }
}
function cleanupAfterDownload(entry) {
// After Sonarr/Radarr hardlink-import, drop the staging copy.
const fpath = entryPath(entry);
let attempts = 0;
const check = setInterval(() => {
attempts++;
try {
if (fs.statSync(fpath).nlink > 1) {
fs.unlinkSync(fpath);
console.log(`[cleanup] auto-deleted after import: ${entry.name}`);
clearInterval(check);
} else if (attempts >= 60) {
clearInterval(check);
}
} catch { clearInterval(check); }
}, 10_000);
}
function dequeue() {
while (activeDownloads < MAX_CONCURRENT && waitingQueue.length > 0) {
const { entry, webseedUrl } = waitingQueue.shift();
// Skip entries cancelled while sitting in the wait list
if (isCancelled(entry.hash) || !downloads.has(entry.hash)) {
cancelled.delete(entry.hash);
continue;
}
activeDownloads++;
entry.state = 'downloading';
saveState();
startDownload(entry, webseedUrl);
}
}
function startDownload(entry, webseedUrl) {
(async () => {
try {
if (isCancelled(entry.hash) || !downloads.has(entry.hash)) {
throw new Error('cancelled');
}
console.log(`[qbt] downloading ${entry.name} via ${webseedUrl.slice(0, 60)}…`);
await streamToFile(webseedUrl, entry);
if (isCancelled(entry.hash) || !downloads.has(entry.hash)) {
throw new Error('cancelled');
}
// Use actual file size — Content-Length header can lie
try { entry.size = fs.statSync(entryPath(entry)).size; } catch {}
entry.progress = 1.0;
entry.state = 'uploading';
fixOwnership(entryPath(entry));
console.log(`[qbt] done: ${entryPath(entry)}`);
cleanupAfterDownload(entry);
} catch (err) {
const msg = err && err.message ? err.message : String(err);
// Delete partial file before retry / after cancel
try { fs.unlinkSync(entryPath(entry)); } catch { }
if (isCancelled(entry.hash) || msg === 'cancelled') {
console.log(`[qbt] cancelled: ${entry.name}`);
} else {
// "File temporarily unavailable" is Webshare's FATAL error — not actually retryable.
const retryable = msg.includes('stalled') || msg.includes('ECONNRESET')
|| msg.includes('HTTP 5');
if (retryable && downloads.has(entry.hash)) {
console.warn(`[qbt] will retry ${entry.name} in 2 min: ${msg}`);
entry.state = 'queuedDL';
saveState();
setTimeout(() => {
if (!isCancelled(entry.hash) && downloads.has(entry.hash)) {
enqueue(entry, webseedUrl);
}
}, 2 * 60 * 1000);
} else if (downloads.has(entry.hash)) {
entry.state = 'error';
console.error(`[qbt] failed ${entry.name}: ${msg}`);
// Sonarr treats qBittorrent's "error" state as an advisory warning, not
// a failure — it never auto-blocklists/redownloads on its own. Removing
// the entry makes the download disappear from the client's list, which
// Sonarr does treat as failed (triggering its existing auto-redownload).
setTimeout(() => {
downloads.delete(entry.hash);
saveState();
}, 3 * 60 * 1000);
}
}
} finally {
cancelled.delete(entry.hash);
activeRequests.delete(entry.hash);
activeDownloads--;
if (activeDownloads < 0) activeDownloads = 0;
saveState();
dequeue();
}
})();
}
function enqueue(entry, webseedUrl) {
entry.state = 'queuedDL';
entry.webseedUrl = webseedUrl;
waitingQueue.push({ entry, webseedUrl });
console.log(`[qbt] queued (${waitingQueue.length} waiting, ${activeDownloads}/${MAX_CONCURRENT} active): ${entry.name}`);
saveState();
dequeue();
}
function identToHash(ident) {
return crypto.createHash('sha1').update('ws:' + ident).digest('hex');
}
function streamToFile(url, entry) {
return new Promise((resolve, reject) => {
const follow = (u, hops) => {
if (hops > 10) return reject(new Error('Too many redirects'));
const mod = u.startsWith('https') ? https : http;
const req = mod.get(u, res => {
if (res.statusCode === 301 || res.statusCode === 302 || res.statusCode === 303) {
res.resume();
return follow(res.headers.location, hops + 1);
}
if (res.statusCode !== 200) {
res.resume();
return reject(new Error(`HTTP ${res.statusCode}`));
}
if (isCancelled(entry.hash) || !downloads.has(entry.hash)) {
res.resume();
return reject(new Error('cancelled'));
}
activeRequests.set(entry.hash, req);
entry.size = parseInt(res.headers['content-length'] || '0', 10);
// try to grab filename from Content-Disposition
const cd = res.headers['content-disposition'] || '';
const m = cd.match(/filename\*?=(?:UTF-8'')?["']?([^"';\r\n]+)/i);
if (m) {
const fn = decodeURIComponent(m[1].trim().replace(/^["']|["']$/g, ''));
if (fn) entry.name = safeFileName(fn);
}
let received = 0;
let lastDataAt = Date.now();
ensureDir(entryDir(entry));
const dest = entryPath(entry);
const file = fs.createWriteStream(dest);
// Poll every 10s — avoids setTimeout/clearTimeout on every chunk
const stallCheck = setInterval(() => {
if (isCancelled(entry.hash) || !downloads.has(entry.hash)) {
res.destroy(new Error('cancelled'));
return;
}
if (Date.now() - lastDataAt > 60_000)
res.destroy(new Error('Download stalled (60s no data)'));
}, 10_000);
let speedAccum = 0;
let speedSampleAt = Date.now();
res.on('data', chunk => {
if (isCancelled(entry.hash) || !downloads.has(entry.hash)) {
res.destroy(new Error('cancelled'));
return;
}
received += chunk.length;
lastDataAt = Date.now();
if (entry.size > 0) entry.progress = received / entry.size;
speedAccum += chunk.length;
const elapsed = Date.now() - speedSampleAt;
if (elapsed >= 2000) {
const bps = speedAccum / (elapsed / 1000);
entry.speed = formatSpeed(bps);
speedAccum = 0;
speedSampleAt = Date.now();
}
});
res.pipe(file);
const done = () => {
clearInterval(stallCheck);
if (activeRequests.get(entry.hash) === req) {
activeRequests.delete(entry.hash);
}
};
file.on('finish', () => {
done();
if (isCancelled(entry.hash) || !downloads.has(entry.hash)) {
try { fs.unlinkSync(dest); } catch { /* ignore */ }
return reject(new Error('cancelled'));
}
file.close(resolve);
});
file.on('error', (e) => { done(); reject(e); });
res.on('error', (e) => { done(); reject(e); });
req.on('error', (e) => { done(); reject(e); });
});
req.on('error', reject);
};
follow(url, 0);
});
}
function register(app) {
ensureDir(SAVE_PATH);
// ── auth ──────────────────────────────────────────────────────────────
app.post('/api/v2/auth/login', (req, res) => {
res.cookie('SID', 'webshare-session', { httpOnly: true });
res.send('Ok.');
});
app.get('/api/v2/auth/logout', (req, res) => res.send('Ok.'));
// ── app info ──────────────────────────────────────────────────────────
app.get('/api/v2/app/version', (req, res) => res.send('5.0.0'));
app.get('/api/v2/app/webapiVersion', (req, res) => res.send('2.8.3'));
app.get('/api/v2/app/buildInfo', (req, res) => res.json({
bitness: 64, boost: '1.84.0', libtorrent: '2.0.10', openssl: '3.2.1', qt: '6.6.2', zlib: '1.3.1',
}));
app.get('/api/v2/app/preferences', (req, res) => res.json({
save_path: SAVE_PATH + '/',
temp_path_enabled: false,
create_subfolder_enabled: false,
incomplete_files_ext: false,
}));
// ── torrent list ──────────────────────────────────────────────────────
app.get('/api/v2/torrents/info', (req, res) => {
const list = [...downloads.values()].map(d => ({
hash: d.hash,
name: d.name,
state: d.state,
progress: d.progress,
size: d.size,
downloaded: Math.floor(d.progress * d.size),
save_path: entryDir(d) + '/',
content_path: entryPath(d),
num_seeds: 0,
num_leechs: 0,
ratio: 0,
eta: (d.state === 'downloading' || d.state === 'queuedDL') ? 9999 : 0,
category: d.category || '',
tags: '',
}));
res.json(list);
});
app.get('/api/v2/torrents/properties', (req, res) => {
const d = downloads.get(req.query.hash);
if (!d) return res.status(404).json({});
res.json({ save_path: SAVE_PATH + '/', hash: d.hash });
});
// Sonarr calls this to discover the file list inside a torrent.
// Without it Sonarr falls back to directory-scan mode and reports
// "No files found are eligible for import in /path/to/file.mkv".
app.get('/api/v2/torrents/files', (req, res) => {
const d = downloads.get(req.query.hash);
if (!d) return res.status(404).json({});
res.json([{
name: d.name,
size: d.size,
progress: d.progress,
priority: 1,
is_seed: d.state === 'uploading',
piece_range: [0, 0],
availability: d.state === 'uploading' ? 1 : -1,
}]);
});
app.get('/api/v2/torrents/categories', (req, res) => res.json({
webshare: { name: 'webshare', savePath: SAVE_PATH + '/' },
}));
// ── add download — accepts both url list AND .torrent file upload ─────
app.post('/api/v2/torrents/add', upload.fields([{ name: 'torrents' }]), (req, res) => {
res.send('Ok.');
const body = req.body || {};
const category = body.category || '';
// Case 1: Sonarr sends a .torrent file (multipart upload)
const files = req.files && req.files['torrents'];
if (files && files.length) {
for (const f of files) {
const webseedUrl = extractWebseed(f.buffer);
const hash = extractInfoHash(f.buffer)
|| crypto.createHash('sha1').update(f.buffer).digest('hex');
if (!webseedUrl) {
// Real multi-file torrent — we can't download it.
// Record as 'error' so Sonarr stops re-adding this release.
if (!downloads.has(hash)) {
const name = f.originalname.replace(/\.torrent$/i, '') || hash;
downloads.set(hash, { hash, name, state: 'error', progress: 0, size: 0, category });
saveState();
console.warn(`[qbt] no url-list, recording as error: ${name}`);
}
continue;
}
// If we already have this webseed URL under any hash (e.g. old sha1(url) hash),
// re-map it to the correct info hash so Sonarr can track it.
const existingByUrl = [...downloads.values()]
.find(e => e.webseedUrl === webseedUrl);
if (existingByUrl) {
if (existingByUrl.hash !== hash) {
downloads.delete(existingByUrl.hash);
existingByUrl.hash = hash;
downloads.set(hash, existingByUrl);
saveState();
console.log(`[qbt] re-hashed existing entry: ${existingByUrl.name}`);
}
continue;
}
if (downloads.has(hash)) continue;
const name = f.originalname.replace(/\.torrent$/i, '') || hash;
const entry = { hash, name, state: 'downloading', progress: 0, size: 0, category };
downloads.set(hash, entry);
console.log(`[qbt] queued via .torrent: ${name}`);
enqueue(entry, webseedUrl);
}
return;
}
// Case 2: url list (our old path, kept as fallback)
const urls = (body.urls || '').trim().split(/\s+/).filter(Boolean);
for (const url of urls) {
const m = url.match(/\/(?:download|stream)\/([^/?&#]+)/);
const ident = m ? m[1] : null;
const hash = crypto.createHash('sha1').update(url).digest('hex');
if (downloads.has(hash)) continue;
const entry = { hash, name: (ident || hash), state: 'downloading', progress: 0, size: 0, category };
downloads.set(hash, entry);
console.log(`[qbt] queued via url: ${url}`);
enqueue(entry, url);
}
});
// ── cleanup: delete staging files that have been imported or are stale ───
// Fast path: Sonarr hardlinks → nlink > 1 → delete immediately.
// Fallback: Sonarr copies instead of hardlinks (nlink stays 1) → delete
// any completed file whose mtime is older than 2 hours. Never touch files
// that are actively downloading or queued.
function cleanupStaging() {
let staged;
try { staged = fs.readdirSync(SAVE_PATH); } catch { return; }
const now = Date.now();
const TWO_HOURS = 2 * 60 * 60 * 1000;
const activeFiles = new Set(
[...downloads.values()]
.filter(e => e.state === 'downloading' || e.state === 'queuedDL')
.map(e => e.name)
);
for (const fname of staged) {
if (fname.startsWith('.')) continue;
if (activeFiles.has(fname)) continue;
const fpath = path.join(SAVE_PATH, fname);
try {
const st = fs.statSync(fpath);
if (!st.isFile()) continue;
if (st.nlink > 1) {
fs.unlinkSync(fpath);
console.log(`[cleanup] deleted after hardlink: ${fname}`);
} else if (now - st.mtimeMs > TWO_HOURS) {
fs.unlinkSync(fpath);
console.log(`[cleanup] deleted stale staging file (>2h): ${fname}`);
}
} catch { }
}
}
// Load persisted state on startup, then run cleanup
loadState();
cleanupStaging();
setInterval(cleanupStaging, 15 * 60 * 1000);
// ── control ───────────────────────────────────────────────────────────
app.post('/api/v2/torrents/delete', (req, res) => {
const hashes = (req.body.hashes || '').split('|').filter(Boolean);
for (const h of hashes) {
// Same path as UI cancel: abort stream, free concurrent slot, start next queued
cancelDownload(h);
}
res.send('Ok.');
});
app.post('/api/v2/torrents/pause', (req, res) => res.send('Ok.'));
app.post('/api/v2/torrents/resume', (req, res) => res.send('Ok.'));
app.post('/api/v2/torrents/setCategory', (req, res) => res.send('Ok.'));
app.post('/api/v2/torrents/setLocation', (req, res) => res.send('Ok.'));
app.post('/api/v2/torrents/rename', (req, res) => res.send('Ok.'));
app.post('/api/v2/torrents/createCategory', (req, res) => res.send('Ok.'));
app.post('/api/v2/torrents/editCategory', (req, res) => res.send('Ok.'));
app.post('/api/v2/torrents/removeCategories', (req, res) => res.send('Ok.'));
}
function getQueue() {
return [...downloads.values()].map(e => ({
hash: e.hash,
name: e.name,
size: e.size,
progress: e.progress,
state: e.state,
speed: e.speed || '',
path: entryPath(e),
}));
}
/**
* UI / manual grab: same staging path as *arr (`DOWNLOAD_PATH` / webshare).
* Sonarr/Radarr own library placement (Shows vs Movies) after import.
*/
function enqueueDirect(ident, name, size) {
const hash = identToHash(ident);
if (downloads.has(hash)) return false;
const webseedUrl = `${process.env.BASE_URL || `http://localhost:${process.env.PORT || 3001}`}/stream/${ident}`;
const fileName = safeFileName(name || ident);
ensureDir(SAVE_PATH);
const entry = {
hash,
name: fileName,
size: size || 0,
progress: 0,
state: 'downloading',
category: 'webshare',
speed: '',
saveDir: SAVE_PATH,
};
downloads.set(hash, entry);
console.log(`[qbt] stage target: ${path.join(SAVE_PATH, fileName)}`);
enqueue(entry, webseedUrl);
return true;
}
/**
* Cancel a download by hash (UI / Sonarr delete).
* Aborts any in-flight HTTP stream, drops waiting-queue entries, frees a concurrent
* slot (via the download task's finally), and starts the next queued item.
*/
function cancelDownload(hash) {
if (!hash) return false;
const entry = downloads.get(hash);
const inWaiting = waitingQueue.findIndex(item => item.entry.hash === hash);
const hadActiveReq = activeRequests.has(hash);
const wasTracked = Boolean(entry) || inWaiting !== -1 || hadActiveReq;
if (!wasTracked) return false;
cancelled.add(hash);
// Drop from wait list (may be multiple if re-queued oddly — remove all)
for (let i = waitingQueue.length - 1; i >= 0; i--) {
if (waitingQueue[i].entry.hash === hash) waitingQueue.splice(i, 1);
}
// Abort HTTP stream so the download task rejects and its finally() runs:
// activeDownloads-- + dequeue() → next queued item starts.
const req = activeRequests.get(hash);
if (req) {
try {
req.destroy(new Error('cancelled'));
} catch { /* ignore */ }
activeRequests.delete(hash);
}
const wasInFlight = Boolean(entry && entry.state === 'downloading') || hadActiveReq;
if (entry) {
// Only delete partials for incomplete work — keep finished library files
if (entry.state === 'downloading' || entry.state === 'queuedDL') {
try { fs.unlinkSync(entryPath(entry)); } catch { /* ignore */ }
}
downloads.delete(hash);
console.log(`[qbt] cancel: ${entry.name} (${entry.state})`);
} else {
console.log(`[qbt] cancel: ${hash.slice(0, 12)}…`);
}
saveState();
// Waiting-only cancel never enters startDownload finally — clear flag now.
// In-flight cancel keeps the flag until the task's finally block.
if (!wasInFlight) cancelled.delete(hash);
// Waiting-only cancel does not free an active slot; still try dequeue in case
// capacity is free. Active cancel relies on startDownload finally → dequeue,
// but also nudge immediately in case the request was already idle/gone.
dequeue();
return true;
}
module.exports = { register, getQueue, enqueueDirect, cancelDownload };