Consolidate MangaCircus app, scraper, and Compose baseline

scraper/jobs/weebcentral-queue.js

@@ -0,0 +1,124 @@
+'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 };