'use strict'; const fs = require('fs').promises; const path = require('path'); const { runArchive } = require('./weebcentral-archive'); const { sleep } = require('../lib/delay'); async function exists(file) { try { await fs.access(file); return true; } catch { return false; } } async function writeJsonAtomic(file, value) { const temp = `${file}.tmp`; await fs.writeFile(temp, JSON.stringify(value, null, 2) + '\n'); await fs.rename(temp, file).catch(async error => { if (!['EPERM', 'EEXIST'].includes(error.code)) throw error; await fs.rm(file, { force: true }); await fs.rename(temp, file); }); } async function acquireLock(lockFile) { try { const handle = await fs.open(lockFile, 'wx'); await handle.writeFile(String(process.pid)); await handle.close(); return; } catch (error) { if (error.code !== 'EEXIST') throw error; } const priorPid = Number((await fs.readFile(lockFile, 'utf8').catch(() => '')).trim()); try { if (priorPid) process.kill(priorPid, 0); throw new Error(`Queue runner is already active (PID ${priorPid || 'unknown'})`); } catch (error) { if (error.message.startsWith('Queue runner is already active')) throw error; await fs.rm(lockFile, { force: true }); const handle = await fs.open(lockFile, 'wx'); await handle.writeFile(String(process.pid)); await handle.close(); } } async function runQueue(options) { const libraryRoot = path.resolve(options.libraryRoot || 'library'); const queueRoot = path.join(libraryRoot, '.mcscraper-queue'); const stateFile = path.join(queueRoot, 'queue.json'); const logFile = path.join(queueRoot, 'queue.log'); const lockFile = path.join(queueRoot, 'queue.lock'); await fs.mkdir(queueRoot, { recursive: true }); await acquireLock(lockFile); let state = { version: 1, status: 'waiting', waitForFolder: options.waitForFolder || null, entries: options.entries, completed: [], created: new Date().toISOString(), }; if (await exists(stateFile)) { const saved = JSON.parse(await fs.readFile(stateFile, 'utf8')); state = { ...state, ...saved, entries: saved.entries?.length ? saved.entries : options.entries }; } state.pid = process.pid; const event = async (level, message, details = {}) => { const entry = { time: new Date().toISOString(), level, message, ...details }; await fs.appendFile(logFile, JSON.stringify(entry) + '\n'); state.lastEvent = entry; state.updated = entry.time; if (level === 'error') state.lastError = entry; await writeJsonAtomic(stateFile, state); }; const cleanup = async () => { await fs.rm(lockFile, { force: true }).catch(() => {}); }; process.once('exit', () => {}); try { await event('info', `Queue runner started (PID ${process.pid})`); if (state.waitForFolder) { const prerequisite = path.join(libraryRoot, state.waitForFolder); state.status = 'waiting'; await event('info', `Waiting for ${state.waitForFolder} to finish`); while (!(await exists(prerequisite))) await sleep(60000); await event('info', `${state.waitForFolder} is complete; starting queue`); } const completed = new Set(state.completed || []); for (let index = 0; index < state.entries.length; index++) { const item = state.entries[index]; if (completed.has(item.id)) continue; state.status = 'downloading'; state.current = { ...item, position: index + 1, total: state.entries.length }; state.lastError = null; await event('info', `Starting ${item.title}`, { title: item.title }); try { await runArchive({ seriesUrl: item.url, libraryRoot, metadataSearch: item.metadataSearch || item.title, minMs: options.minMs, maxMs: options.maxMs, chapterMinMs: options.chapterMinMs, chapterMaxMs: options.chapterMaxMs, }); } catch (error) { state.status = 'error'; await event('error', error.message, { title: item.title }); throw error; } completed.add(item.id); state.completed = [...completed]; state.current = null; await event('info', `Completed ${item.title}`, { title: item.title }); } state.status = 'complete'; state.current = null; await event('info', 'Queue complete'); } finally { await cleanup(); } } module.exports = { runQueue };