MangaCircus/scraper/jobs/weebcentral-queue.js

'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 };