900 lines
48 KiB
JavaScript
900 lines
48 KiB
JavaScript
'use strict';
|
|
|
|
const fs = require('fs').promises;
|
|
const path = require('path');
|
|
const crypto = require('crypto');
|
|
const { adapterFor, adapters } = require('./adapters');
|
|
const { ArchiveControlError } = require('./lib/control');
|
|
const { sleep } = require('./lib/delay');
|
|
const { normalizeScope, selectByScope, scopeLabel } = require('./lib/scope');
|
|
const { normalizeWorkIdentity, resolveWorkIdentity } = require('./lib/work-identity');
|
|
const { recordSource, recordFailure, sourceKey, ledgerRows } = require('./lib/source-ledger');
|
|
const { scanLibrary, findManga, shouldDownload, isNewerThanLibrary, summarizeInventory, setAutoUpdate } = require('./lib/library-inventory');
|
|
const { inspectLegacyWork, promoteLegacyWork } = require('./lib/legacy-publication');
|
|
const { classifyFailure, safeMessage, progressKey } = require('./lib/failures');
|
|
const { writeJsonAtomic } = require('./lib/json-file');
|
|
|
|
const RATE_KEYS = ['minMs', 'maxMs', 'chapterMinMs', 'chapterMaxMs'];
|
|
const FAST_RATE_SETTINGS = {
|
|
minMs: 1000, maxMs: 4000, chapterMinMs: 5000, chapterMaxMs: 15000,
|
|
};
|
|
const DEFAULT_SOURCE_RATES = {
|
|
weebcentral: { ...FAST_RATE_SETTINGS },
|
|
mangadex: { ...FAST_RATE_SETTINGS },
|
|
mangapill: { ...FAST_RATE_SETTINGS },
|
|
};
|
|
const DEFAULT_SETTINGS = {
|
|
...FAST_RATE_SETTINGS,
|
|
sourceRates: Object.fromEntries(Object.entries(DEFAULT_SOURCE_RATES).map(([source, rate]) => [source, { ...rate }])),
|
|
autoUpdateEnabled: true, autoUpdateTime: '03:00',
|
|
};
|
|
|
|
async function exists(file) {
|
|
try { await fs.access(file); return true; } catch { return false; }
|
|
}
|
|
|
|
function validateRateSettings(input, prior, prefix = '') {
|
|
const next = { ...prior };
|
|
for (const key of RATE_KEYS) {
|
|
if (input[key] === undefined) continue;
|
|
const value = Number(input[key]);
|
|
if (!Number.isFinite(value) || value < 500 || value > 3600000) throw new Error(`${prefix}${key} must be between 500 and 3600000 ms`);
|
|
next[key] = Math.round(value);
|
|
}
|
|
if (next.maxMs < next.minMs) throw new Error(`${prefix}maxMs must be at least ${prefix}minMs`);
|
|
if (next.chapterMaxMs < next.chapterMinMs) throw new Error(`${prefix}chapterMaxMs must be at least ${prefix}chapterMinMs`);
|
|
return next;
|
|
}
|
|
|
|
function validateSettings(input = {}, prior = DEFAULT_SETTINGS) {
|
|
const priorBase = { ...FAST_RATE_SETTINGS, ...prior };
|
|
const next = validateRateSettings(input, priorBase);
|
|
if (input.sourceRates !== undefined && (!input.sourceRates || typeof input.sourceRates !== 'object' || Array.isArray(input.sourceRates))) {
|
|
throw new Error('sourceRates must be an object');
|
|
}
|
|
const requestedRates = input.sourceRates || {};
|
|
const sourceIds = new Set([
|
|
...Object.keys(DEFAULT_SOURCE_RATES),
|
|
...Object.keys(prior.sourceRates || {}),
|
|
...Object.keys(requestedRates),
|
|
]);
|
|
next.sourceRates = {};
|
|
for (const source of sourceIds) {
|
|
if (!/^[a-z0-9][a-z0-9-]*$/.test(source)) throw new Error(`Invalid source id: ${source}`);
|
|
const patch = requestedRates[source];
|
|
if (patch !== undefined && (!patch || typeof patch !== 'object' || Array.isArray(patch))) {
|
|
throw new Error(`sourceRates.${source} must be an object`);
|
|
}
|
|
const base = {
|
|
...FAST_RATE_SETTINGS,
|
|
...(DEFAULT_SOURCE_RATES[source] || {}),
|
|
...((prior.sourceRates || {})[source] || {}),
|
|
};
|
|
next.sourceRates[source] = validateRateSettings(patch || {}, base, `sourceRates.${source}.`);
|
|
}
|
|
if (input.autoUpdateEnabled !== undefined) next.autoUpdateEnabled = !!input.autoUpdateEnabled;
|
|
if (input.autoUpdateTime !== undefined) {
|
|
const value = String(input.autoUpdateTime).trim();
|
|
if (!/^(?:[01]\d|2[0-3]):[0-5]\d$/.test(value)) throw new Error('autoUpdateTime must be HH:MM');
|
|
next.autoUpdateTime = value;
|
|
}
|
|
return next;
|
|
}
|
|
|
|
function settingsForSource(settings, source) {
|
|
return { ...settings, ...((settings.sourceRates || {})[source] || {}) };
|
|
}
|
|
|
|
function existingPolicy(value) {
|
|
const policy = String(value || 'skip-any');
|
|
if (!['skip-any', 'add-source'].includes(policy)) throw new Error('existingPolicy must be skip-any or add-source');
|
|
return policy;
|
|
}
|
|
|
|
function nextSchedule(time, from = new Date()) {
|
|
const [hours, minutes] = String(time).split(':').map(Number);
|
|
const next = new Date(from);
|
|
next.setHours(hours, minutes, 0, 0);
|
|
if (next <= from) next.setDate(next.getDate() + 1);
|
|
return next;
|
|
}
|
|
|
|
class ScraperService {
|
|
constructor(options = {}) {
|
|
this.libraryRoot = path.resolve(options.libraryRoot || 'library');
|
|
this.dataRoot = path.resolve(options.dataRoot || path.join('data', 'scraper'));
|
|
this.stateFile = path.join(this.dataRoot, 'queue.json');
|
|
this.logFile = path.join(this.dataRoot, 'events.log');
|
|
this.lockFile = path.join(this.dataRoot, 'daemon.lock');
|
|
this.legacyStateFile = path.join(this.libraryRoot, '.mcscraper-queue', 'queue.json');
|
|
this.state = null;
|
|
this.stopping = false;
|
|
this.loopPromise = null;
|
|
this.active = new Map();
|
|
this.writeChain = Promise.resolve();
|
|
this.discoveryTimer = null;
|
|
this.inventoryCache = null;
|
|
this.inventoryCachedAt = 0;
|
|
this.inventoryGeneration = 0;
|
|
this.inventoryCacheGeneration = -1;
|
|
this.inventoryRefresh = null;
|
|
this.inventoryAttemptedAt = 0;
|
|
this.inventoryError = null;
|
|
this.scanInventory = options.scanLibrary || scanLibrary;
|
|
this.resolveAdapter = options.adapterFor || adapterFor;
|
|
this.adapterList = options.adapters || adapters;
|
|
this.retryDelayMs = options.retryDelayMs ?? 300000;
|
|
this.runtimeHeartbeat = Date.now();
|
|
}
|
|
|
|
async init() {
|
|
await fs.mkdir(this.dataRoot, { recursive: true });
|
|
await fs.mkdir(this.libraryRoot, { recursive: true });
|
|
await this.acquireLock();
|
|
this.state = await this.loadState();
|
|
this.state.incidents = this.state.incidents || [];
|
|
this.state.circuits = this.state.circuits || {};
|
|
this.state.updateChecks = this.state.updateChecks || {};
|
|
this.state.sourceLedger = this.state.sourceLedger || {};
|
|
// Older releases retained successful no-change checks in the work queue.
|
|
// Keep their latest result before retiring those finished rows.
|
|
for (const item of this.state.queue) {
|
|
if (item.kind !== 'discovery' || item.status !== 'complete' || !item.checked || !item.libraryMangaId) continue;
|
|
const key = JSON.stringify([item.libraryMangaId, item.source]);
|
|
const prior = this.state.updateChecks[key];
|
|
if (!prior?.lastCheckedAt || Date.parse(item.checked) > Date.parse(prior.lastCheckedAt)) {
|
|
this.state.updateChecks[key] = { ...prior, mangaId: item.libraryMangaId, source: item.source,
|
|
lastCheckedAt: item.checked, lastResult: 'no-updates' };
|
|
}
|
|
}
|
|
this.state.queue = this.state.queue.filter(item => item.kind !== 'discovery' || item.status !== 'complete' || !item.libraryMangaId);
|
|
this.state.queue.forEach((item, index) => { item.order = index; });
|
|
for (const item of this.state.queue) {
|
|
delete item.stopReason;
|
|
if (item.status === 'downloading' || item.status === 'paused' || item.status === 'discovering') item.status = 'queued';
|
|
const found = this.resolveAdapter(item.url);
|
|
item.source = found?.adapter.id || item.source || 'unknown';
|
|
item.seriesId = item.seriesId || found?.info.id || item.id;
|
|
item.kind = ['discovery', 'inspection'].includes(item.kind) ? item.kind : 'download';
|
|
item.scope = normalizeScope(item.scope || {});
|
|
item.scopeLabel = item.scopeLabel || scopeLabel(item.scope);
|
|
item.existingPolicy = existingPolicy(item.existingPolicy);
|
|
if (item.status === 'error') {
|
|
item.failure = item.failure || classifyFailure(new Error(item.error));
|
|
if (item.failure.kind === 'unavailable') item.status = 'unavailable';
|
|
}
|
|
if (item.source === 'mangadex' && item.status === 'unavailable' && !item.availability
|
|
&& item.error === 'No downloadable English MangaDex chapters found') {
|
|
item.error = 'The previous adapter selected no numbered hosted English chapters. Inspect this source again to verify its edition and availability, including pilots and specials.';
|
|
if (item.failure) item.failure = { ...item.failure, message: item.error };
|
|
}
|
|
// Existing queue rows are leads. Only a recorded catalog observation may
|
|
// promote one to an adapter-compatible source.
|
|
if (!this.state.sourceLedger[sourceKey(item)]) recordSource(this.state.sourceLedger, item);
|
|
}
|
|
this.state.version = 4;
|
|
this.state.currentId = null;
|
|
this.state.currentIds = {};
|
|
this.state.sourcePaused = this.state.sourcePaused || {};
|
|
// A legacy title-unavailable pause can be released; access and unknown failures remain paused.
|
|
for (const source of Object.keys(this.state.sourcePaused)) {
|
|
const failed = this.state.queue.filter(item => item.source === source && ['error', 'unavailable', 'quarantined'].includes(item.status));
|
|
if (failed.length && failed.every(item => item.failure?.kind === 'unavailable')) delete this.state.sourcePaused[source];
|
|
}
|
|
this.state.status = this.state.paused ? 'paused' : 'idle';
|
|
this.state.pid = process.pid;
|
|
await this.event('info', 'Scraper daemon started', { pid: process.pid });
|
|
this.scheduleDiscovery();
|
|
this.loopPromise = this.workLoop();
|
|
return this;
|
|
}
|
|
|
|
async acquireLock() {
|
|
try {
|
|
const handle = await fs.open(this.lockFile, 'wx');
|
|
await handle.writeFile(String(process.pid));
|
|
return await handle.close();
|
|
} catch (error) {
|
|
if (error.code !== 'EEXIST') throw error;
|
|
}
|
|
const oldPid = Number((await fs.readFile(this.lockFile, 'utf8').catch(() => '')).trim());
|
|
try {
|
|
if (oldPid) process.kill(oldPid, 0);
|
|
throw new Error(`Scraper daemon is already running (PID ${oldPid || 'unknown'})`);
|
|
} catch (error) {
|
|
if (error.code !== 'ESRCH') throw error;
|
|
await fs.rm(this.lockFile, { force: true });
|
|
const handle = await fs.open(this.lockFile, 'wx');
|
|
await handle.writeFile(String(process.pid));
|
|
await handle.close();
|
|
}
|
|
}
|
|
|
|
async loadState() {
|
|
if (await exists(this.stateFile)) {
|
|
const saved = JSON.parse(await fs.readFile(this.stateFile, 'utf8'));
|
|
return {
|
|
version: 3, paused: false, status: 'idle', queue: [], history: [], publishedRevision: 0,
|
|
currentIds: {}, sourcePaused: {},
|
|
settings: DEFAULT_SETTINGS, ...saved,
|
|
settings: validateSettings(saved.settings || {}, DEFAULT_SETTINGS),
|
|
};
|
|
}
|
|
const state = {
|
|
version: 3, paused: false, status: 'idle', currentId: null, currentIds: {}, sourcePaused: {},
|
|
settings: { ...DEFAULT_SETTINGS }, queue: [], history: [], publishedRevision: 0,
|
|
created: new Date().toISOString(), lastError: null,
|
|
};
|
|
if (await exists(this.legacyStateFile)) {
|
|
const legacy = JSON.parse(await fs.readFile(this.legacyStateFile, 'utf8'));
|
|
const completed = new Set(legacy.completed || []);
|
|
state.queue = (legacy.entries || []).map((item, index) => ({
|
|
id: item.id, seriesId: item.id, source: 'weebcentral', title: item.title, url: item.url,
|
|
metadataSearch: item.metadataSearch || item.title,
|
|
kind: 'download', scope: { mode: 'missing' }, existingPolicy: 'skip-any',
|
|
status: completed.has(item.id) ? 'complete' : 'queued', order: index,
|
|
added: legacy.created || new Date().toISOString(),
|
|
}));
|
|
state.settings = validateSettings(legacy.settings || {}, DEFAULT_SETTINGS);
|
|
state.migratedFrom = this.legacyStateFile;
|
|
}
|
|
return state;
|
|
}
|
|
|
|
async persist() {
|
|
const write = async () => {
|
|
this.state.updated = new Date().toISOString();
|
|
await writeJsonAtomic(this.stateFile, this.state);
|
|
};
|
|
const pending = this.writeChain.then(write, write);
|
|
this.writeChain = pending.catch(() => {});
|
|
return pending;
|
|
}
|
|
|
|
async event(level, message, details = {}) {
|
|
const entry = { time: new Date().toISOString(), level, message: safeMessage(message), ...details };
|
|
if (entry.url) entry.url = safeMessage(entry.url);
|
|
await fs.appendFile(this.logFile, JSON.stringify(entry) + '\n');
|
|
console.log(JSON.stringify({ service: 'scraper', ...entry }));
|
|
this.state.lastEvent = entry;
|
|
if (level === 'error') this.state.lastError = entry;
|
|
await this.persist();
|
|
}
|
|
|
|
async recentEvents(limit = 30) {
|
|
const handle = await fs.open(this.logFile, 'r').catch(error => {
|
|
if (error.code === 'ENOENT') return null;
|
|
throw error;
|
|
});
|
|
if (!handle) return [];
|
|
let text;
|
|
try {
|
|
const { size } = await handle.stat();
|
|
const length = Math.min(size, 512 * 1024), start = size - length;
|
|
const buffer = Buffer.alloc(length);
|
|
const { bytesRead } = await handle.read(buffer, 0, length, start);
|
|
text = buffer.subarray(0, bytesRead).toString('utf8');
|
|
if (start > 0) {
|
|
const boundary = text.indexOf('\n');
|
|
text = boundary < 0 ? '' : text.slice(boundary + 1);
|
|
}
|
|
} finally { await handle.close(); }
|
|
return text.trim().split(/\r?\n/).filter(Boolean).slice(-Math.max(1, Math.min(200, limit)))
|
|
.map(line => { try { return JSON.parse(line); } catch { return { level: 'warning', message: line }; } });
|
|
}
|
|
|
|
invalidateInventory() {
|
|
// Keep the last summary visible while rebuilding. Download/discovery decisions
|
|
// still await an authoritative scan through libraryInventory(), never this stale view.
|
|
this.inventoryGeneration++;
|
|
}
|
|
|
|
refreshInventory() {
|
|
if (this.inventoryRefresh) return this.inventoryRefresh;
|
|
const generation = this.inventoryGeneration;
|
|
this.inventoryAttemptedAt = Date.now();
|
|
this.inventoryRefresh = Promise.resolve().then(() => this.scanInventory(this.libraryRoot))
|
|
.then(async inventory => {
|
|
// A chapter or metadata edit during the scan invalidates its result.
|
|
if (generation === this.inventoryGeneration) {
|
|
this.inventoryCache = inventory;
|
|
this.inventoryCachedAt = Date.now();
|
|
this.inventoryCacheGeneration = generation;
|
|
}
|
|
// Backfill source leads from existing library metadata without probing
|
|
// hosts or asserting that their current catalog is complete.
|
|
let added = 0;
|
|
if (generation === this.inventoryGeneration) {
|
|
for (const manga of inventory.mangas || []) {
|
|
for (const source of manga.sources || []) {
|
|
const found = this.resolveAdapter(source.url);
|
|
if (!found) continue;
|
|
const lead = { source: found.adapter.id, seriesId: found.info.id, url: source.url,
|
|
title: manga.title, libraryMangaId: manga.id,
|
|
workIdentity: resolveWorkIdentity(manga.meta, manga.id) };
|
|
const existing = this.state.sourceLedger[sourceKey(lead)];
|
|
if (existing && !existing.workKey?.startsWith('source:')) continue;
|
|
recordSource(this.state.sourceLedger, lead);
|
|
added++;
|
|
}
|
|
}
|
|
}
|
|
if (added) await this.persist();
|
|
this.inventoryError = null;
|
|
return inventory;
|
|
}).catch(error => {
|
|
this.inventoryError = safeMessage(error.message);
|
|
throw error;
|
|
}).finally(() => { this.inventoryRefresh = null; });
|
|
return this.inventoryRefresh;
|
|
}
|
|
|
|
async libraryInventory(force = false) {
|
|
if (!force && this.inventoryCache && this.inventoryCacheGeneration === this.inventoryGeneration
|
|
&& Date.now() - this.inventoryCachedAt < 60000) return this.inventoryCache;
|
|
do {
|
|
await this.refreshInventory();
|
|
} while (this.inventoryCacheGeneration !== this.inventoryGeneration);
|
|
return this.inventoryCache;
|
|
}
|
|
|
|
scheduleDiscovery() {
|
|
if (this.discoveryTimer) clearTimeout(this.discoveryTimer);
|
|
this.discoveryTimer = null;
|
|
if (!this.state?.settings?.autoUpdateEnabled || this.stopping) {
|
|
if (this.state) this.state.nextDiscoveryAt = null;
|
|
return;
|
|
}
|
|
const next = nextSchedule(this.state.settings.autoUpdateTime);
|
|
this.state.nextDiscoveryAt = next.toISOString();
|
|
const delay = Math.max(1000, next.getTime() - Date.now());
|
|
this.discoveryTimer = setTimeout(() => {
|
|
this.discoveryTimer = null;
|
|
this.enqueueDiscoveryChecks('schedule').catch(error => this.event('error', `Nightly discovery failed: ${error.message}`))
|
|
.finally(() => { if (!this.stopping) this.scheduleDiscovery(); });
|
|
}, delay);
|
|
this.discoveryTimer.unref();
|
|
}
|
|
|
|
pendingSeries(source, seriesId) {
|
|
return this.state.queue.some(item => item.source === source && String(item.seriesId) === String(seriesId)
|
|
&& ['queued', 'downloading', 'discovering', 'ready', 'error', 'retry_wait', 'quarantined', 'unavailable'].includes(item.status));
|
|
}
|
|
|
|
recordUpdateCheck(item, patch) {
|
|
if (!item.libraryMangaId || !item.source) return;
|
|
const key = JSON.stringify([item.libraryMangaId, item.source]);
|
|
this.state.updateChecks[key] = { ...(this.state.updateChecks[key] || {}),
|
|
mangaId: item.libraryMangaId, source: item.source, ...patch };
|
|
}
|
|
|
|
async enqueueDiscoveryChecks(reason = 'manual') {
|
|
const inventory = await this.libraryInventory(true);
|
|
let queued = 0, skipped = 0;
|
|
for (const manga of inventory.mangas) {
|
|
if (!manga.autoUpdate) continue;
|
|
const supported = manga.sources.map(source => ({ source, found: this.resolveAdapter(source.url) })).filter(item => item.found);
|
|
if (!supported.length) { skipped++; continue; }
|
|
const preferred = manga.meta?.['x-scraper']?.preferredSource;
|
|
const selected = supported.find(item => item.found.adapter.id === preferred) || supported[0];
|
|
const source = selected.found.adapter.id, seriesId = selected.found.info.id;
|
|
if (this.pendingSeries(source, seriesId)) { skipped++; continue; }
|
|
const item = {
|
|
id: crypto.randomUUID(), seriesId, source, kind: 'discovery',
|
|
title: manga.title, url: selected.found.info.url, metadataSearch: manga.title,
|
|
libraryMangaId: manga.id, scope: { mode: 'missing' }, existingPolicy: 'skip-any',
|
|
status: 'queued', order: this.state.queue.length, added: new Date().toISOString(), reason,
|
|
};
|
|
this.state.queue.push(item); queued++;
|
|
}
|
|
this.state.lastDiscovery = { time: new Date().toISOString(), reason, queued, skipped };
|
|
await this.event('info', `Queued ${queued} update check${queued === 1 ? '' : 's'}`, { reason, queued, skipped });
|
|
return { queued, skipped };
|
|
}
|
|
|
|
async setMangaAutoUpdate(id, enabled) {
|
|
const result = await setAutoUpdate(this.libraryRoot, id, enabled);
|
|
this.invalidateInventory();
|
|
await this.event('info', `${enabled ? 'Enabled' : 'Disabled'} auto update for ${id}`, { mangaId: id, enabled: !!enabled });
|
|
return result;
|
|
}
|
|
|
|
async publishLegacyReady() {
|
|
if (!this.state.paused || this.active.size || Object.keys(this.state.currentIds || {}).length) {
|
|
throw new Error('Pause the queue and wait for active workers to stop before publishing legacy work');
|
|
}
|
|
const candidates = [];
|
|
const seen = new Set();
|
|
for (const item of this.state.queue) {
|
|
const seriesId = String(item.seriesId || item.id || '');
|
|
if (!seriesId || seen.has(seriesId) || item.status === 'complete') continue;
|
|
seen.add(seriesId);
|
|
const inspection = await inspectLegacyWork({ libraryRoot: this.libraryRoot, seriesId });
|
|
if (inspection) candidates.push({ item, inspection });
|
|
}
|
|
const published = [];
|
|
for (const candidate of candidates) {
|
|
const result = await promoteLegacyWork(candidate.inspection);
|
|
candidate.item.libraryMangaId = result.mangaId;
|
|
candidate.item.legacyPromotedAt = new Date().toISOString();
|
|
published.push(result);
|
|
this.state.publishedRevision = Number(this.state.publishedRevision || 0) + 1;
|
|
this.invalidateInventory();
|
|
await this.event('info', `Published ${result.chaptersPublished} completed chapters for ${candidate.item.title}`, {
|
|
id: candidate.item.id, seriesId: candidate.item.seriesId, title: candidate.item.title,
|
|
mangaId: result.mangaId, chaptersPublished: result.chaptersPublished,
|
|
});
|
|
}
|
|
return { published };
|
|
}
|
|
|
|
async runDiscovery(item, found, source) {
|
|
if (this.stopping || this.state.paused) throw new ArchiveControlError(this.stopping ? 'stopped' : 'paused');
|
|
item.status = 'discovering';
|
|
const catalog = await found.adapter.discover(item, {
|
|
settings: settingsForSource(this.state.settings, source),
|
|
shouldStop: () => this.stopping ? 'stopped' : this.state.paused ? 'paused' : item.stopReason || false,
|
|
onEvent: event => this.event(event.level, event.message, { id: item.id, title: item.title, source, ...(event.status ? { status: event.status } : {}) }),
|
|
});
|
|
if (this.stopping || this.state.paused || item.stopReason) throw new ArchiveControlError(item.stopReason || (this.stopping ? 'stopped' : 'paused'));
|
|
await this.recordCatalog(item, catalog);
|
|
if (item.kind === 'inspection') {
|
|
if (!catalog.chapters?.length) throw Object.assign(new Error(catalog.availability?.message || 'This source edition has no downloadable chapters'), {
|
|
failureKind: catalog.availability?.reason === 'unrecognized-labels' ? 'adapter' : 'unavailable',
|
|
});
|
|
if (item.inspectionFromComplete) {
|
|
const inventory = await this.libraryInventory(true);
|
|
const manga = findManga(inventory, {
|
|
source, seriesId: item.seriesId, sourceUrl: item.url,
|
|
anilist: catalog.links?.al, libraryMangaId: item.libraryMangaId,
|
|
});
|
|
const missing = selectByScope(catalog.chapters, item.scope)
|
|
.filter(ref => shouldDownload(manga, ref, source, item.existingPolicy));
|
|
if (!missing.length) {
|
|
item.kind = 'download'; item.status = 'complete'; item.checked = new Date().toISOString();
|
|
delete item.inspectionFromComplete;
|
|
for (const incident of this.state.incidents) if (incident.jobId === item.id && incident.state === 'open') {
|
|
incident.state = 'resolved'; incident.resolution = 'inspected-complete';
|
|
}
|
|
await this.event('info', `Source inspected: ${item.sourceTitle}; completed download is already in the library`, { id: item.id, source });
|
|
return;
|
|
}
|
|
}
|
|
delete item.inspectionFromComplete;
|
|
item.status = 'ready'; item.checked = new Date().toISOString();
|
|
for (const incident of this.state.incidents) if (incident.jobId === item.id && incident.state === 'open') {
|
|
incident.state = 'resolved'; incident.resolution = 'inspected-ready';
|
|
}
|
|
await this.event('info', `Source inspected: ${item.sourceTitle}; awaiting download confirmation`, { id: item.id, source });
|
|
return;
|
|
}
|
|
const inventory = await this.libraryInventory(true);
|
|
if (this.stopping || this.state.paused) throw new ArchiveControlError(this.stopping ? 'stopped' : 'paused');
|
|
const manga = findManga(inventory, {
|
|
source, seriesId: item.seriesId, sourceUrl: item.url,
|
|
anilist: catalog.links?.al, libraryMangaId: item.libraryMangaId,
|
|
});
|
|
if (item.libraryMangaId && !manga) throw new Error(`Library manga target not found: ${item.libraryMangaId}`);
|
|
const candidates = selectByScope(catalog.chapters || [], item.scope)
|
|
.filter(ref => shouldDownload(manga, ref, source, item.existingPolicy))
|
|
.filter(ref => isNewerThanLibrary(manga, ref));
|
|
item.checked = new Date().toISOString();
|
|
item.catalogTotal = catalog.chapters?.length || 0;
|
|
this.recordUpdateCheck(item, { lastCheckedAt: item.checked,
|
|
...(candidates.length ? { lastNewContentAt: item.checked, lastResult: 'new-content' } : { lastResult: 'no-updates' }) });
|
|
if (!candidates.length) {
|
|
item.status = 'complete'; item.completed = item.checked; item.discovered = 0;
|
|
this.state.history.push({ id: item.id, title: item.title, source, status: 'checked', time: item.completed, discovered: 0 });
|
|
this.state.queue = this.state.queue.filter(entry => entry.id !== item.id);
|
|
this.state.queue.forEach((entry, index) => { entry.order = index; });
|
|
await this.event('info', `No updates found for ${item.title}`, { id: item.id, title: item.title, source });
|
|
return;
|
|
}
|
|
item.kind = 'download'; item.status = 'queued'; item.discovered = candidates.length;
|
|
item.scope = normalizeScope({ mode: 'ids', ids: candidates.map(ref => ref.id) });
|
|
item.scopeLabel = scopeLabel(item.scope);
|
|
await this.event('info', `Found ${candidates.length} new chapter${candidates.length === 1 ? '' : 's'} for ${item.title}`, {
|
|
id: item.id, title: item.title, source, discovered: candidates.length,
|
|
});
|
|
}
|
|
|
|
async recordCatalog(item, catalog) {
|
|
const title = String(catalog.title || item.title).slice(0, 200);
|
|
const changedEdition = item.confirmedSourceTitle && item.confirmedSourceTitle !== title;
|
|
item.sourceTitle = title;
|
|
item.catalogTotal = catalog.chapters?.length || 0;
|
|
item.availability = catalog.availability || { selectedChapters: item.catalogTotal, coverage: 'unknown' };
|
|
recordSource(this.state.sourceLedger ||= {}, item, {
|
|
catalogTotal: item.catalogTotal, availability: item.availability,
|
|
sourceTitle: title, observedAt: new Date().toISOString(),
|
|
});
|
|
await this.persist();
|
|
if (changedEdition) throw Object.assign(new Error('The source edition title changed since inspection. Inspect and confirm it again before downloading.'), { failureKind: 'unavailable' });
|
|
}
|
|
|
|
async jobState(item) {
|
|
if (!item) return null;
|
|
if (item.kind === 'inspection') return { status: 'discovering', updated: item.started, completed: 0, total: null, current: null };
|
|
const sourceId = item.seriesId || item.id;
|
|
const workName = item.source === 'mangapill'
|
|
? `.mcscraper-work-mangapill-${sourceId}`
|
|
: `.mcscraper-work-${sourceId}`;
|
|
const file = path.join(this.libraryRoot, workName, 'job.json');
|
|
try {
|
|
const job = JSON.parse(await fs.readFile(file, 'utf8'));
|
|
return {
|
|
status: job.status || 'unknown', completed: job.publicationMode === 'incremental' ? Number(job.jobCompleted || 0) : job.completed?.length || 0, total: job.total || null,
|
|
percent: job.total ? Math.round((job.publicationMode === 'incremental' ? Number(job.jobCompleted || 0) : job.completed?.length || 0) * 1000 / job.total) / 10 : null,
|
|
lastChapter: job.lastChapter || null, current: job.current || null,
|
|
lastProblem: job.lastProblem || null, lastError: job.lastError || null,
|
|
problemCount: job.problemCount || 0, rate: job.rate || null, updated: job.updated || null,
|
|
};
|
|
} catch { return null; }
|
|
}
|
|
|
|
async publicState() {
|
|
const activeItems = this.state.queue.filter(item => Object.values(this.state.currentIds || {}).includes(item.id));
|
|
const active = await Promise.all(activeItems.map(async current => ({
|
|
source: this.resolveAdapter(current.url)?.adapter.id || current.source,
|
|
current, job: await this.jobState(current),
|
|
})));
|
|
const current = active[0]?.current || null;
|
|
const job = active[0]?.job || null;
|
|
// Status and watchdog probes must not wait for thousands of bind-mount reads.
|
|
// Share one background scan, serve the previous summary, and surface its age/error.
|
|
if (!this.inventoryRefresh && (this.inventoryCacheGeneration !== this.inventoryGeneration || Date.now() - this.inventoryCachedAt >= 60000)
|
|
&& Date.now() - this.inventoryAttemptedAt >= 10000) {
|
|
this.refreshInventory().catch(() => {});
|
|
}
|
|
const inventory = this.inventoryCache;
|
|
return {
|
|
version: this.state.version, status: this.state.status, paused: this.state.paused,
|
|
pid: this.state.pid, settings: this.state.settings, publishedRevision: this.state.publishedRevision || 0,
|
|
current, job, active, sourcePaused: this.state.sourcePaused || {}, queue: this.state.queue,
|
|
history: (this.state.history || []).slice(-20).reverse(), lastError: this.state.lastError || null,
|
|
lastEvent: this.state.lastEvent || null, updated: this.state.updated || null,
|
|
nextDiscoveryAt: this.state.nextDiscoveryAt || null, lastDiscovery: this.state.lastDiscovery || null,
|
|
library: inventory ? summarizeInventory(inventory, value => this.resolveAdapter(value)) : [],
|
|
updateChecks: Object.values(this.state.updateChecks || {}),
|
|
libraryStatus: {
|
|
ready: !!inventory, refreshing: !!this.inventoryRefresh,
|
|
stale: this.inventoryCacheGeneration !== this.inventoryGeneration || Date.now() - this.inventoryCachedAt >= 60000,
|
|
updated: this.inventoryCachedAt ? new Date(this.inventoryCachedAt).toISOString() : null,
|
|
error: this.inventoryError,
|
|
},
|
|
adapters: this.adapterList.map(item => ({ id: item.id, label: item.label })),
|
|
sourceLedger: ledgerRows(this.state.sourceLedger),
|
|
events: await this.recentEvents(25),
|
|
incidents: (this.state.incidents || []).slice(-50).reverse(), circuits: this.state.circuits || {},
|
|
runtimeHeartbeat: new Date(this.runtimeHeartbeat).toISOString(),
|
|
};
|
|
}
|
|
|
|
async enqueue(input) {
|
|
const found = this.resolveAdapter(input.url);
|
|
if (!found) throw new Error('No installed scraper adapter accepts this URL');
|
|
const workIdentity = normalizeWorkIdentity(input.workIdentity);
|
|
if (input.workIdentity != null && !workIdentity) throw new Error('Invalid manga work identity');
|
|
const scope = normalizeScope(input.scope || {});
|
|
const policy = existingPolicy(input.existingPolicy);
|
|
const signature = JSON.stringify([found.adapter.id, found.info.id, scope, policy, input.libraryMangaId || null]);
|
|
if (this.state.queue.some(item => item.status !== 'complete'
|
|
&& JSON.stringify([item.source, item.seriesId, item.scope, item.existingPolicy, item.libraryMangaId || null]) === signature)) {
|
|
throw new Error('That exact manga scope is already queued');
|
|
}
|
|
const item = {
|
|
...found.info, id: crypto.randomUUID(), seriesId: found.info.id, kind: input.inspect === true ? 'inspection' : 'download',
|
|
title: String(input.title || found.info.title).trim().slice(0, 200),
|
|
metadataSearch: String(input.metadataSearch || input.title || found.info.title).trim().slice(0, 200),
|
|
metadataSearchExplicit: !!String(input.metadataSearch || '').trim(),
|
|
scope, scopeLabel: scopeLabel(scope), existingPolicy: policy,
|
|
...(workIdentity ? { workIdentity } : {}),
|
|
...(input.libraryMangaId ? { libraryMangaId: String(input.libraryMangaId) } : {}),
|
|
status: 'queued', order: this.state.queue.length, added: new Date().toISOString(),
|
|
};
|
|
this.state.queue.push(item);
|
|
recordSource(this.state.sourceLedger ||= {}, item);
|
|
await this.event('info', `Queued ${item.title}: ${item.scopeLabel}`, { id: item.id, seriesId: item.seriesId, title: item.title });
|
|
return item;
|
|
}
|
|
|
|
async remove(id) {
|
|
const item = this.state.queue.find(entry => entry.id === id);
|
|
if (!item) throw new Error('Queue item not found');
|
|
if (Object.values(this.state.currentIds || {}).includes(item.id)) throw new Error('Pause the active download before removing it');
|
|
this.state.queue = this.state.queue.filter(entry => entry.id !== id);
|
|
this.state.queue.forEach((entry, index) => { entry.order = index; });
|
|
await this.event('info', `Removed ${item.title} from queue`, { id: item.id, title: item.title });
|
|
}
|
|
|
|
async clearCompleted(source = '') {
|
|
if (source && !this.adapterList.some(adapter => adapter.id === source)) throw new Error('Unknown source');
|
|
const before = this.state.queue.length;
|
|
const activeIds = new Set(Object.values(this.state.currentIds || {}));
|
|
this.state.queue = this.state.queue.filter(item => !(item.status === 'complete' && item.kind === 'download'
|
|
&& !activeIds.has(item.id) && (!source || item.source === source)));
|
|
this.state.queue.forEach((item, index) => { item.order = index; });
|
|
const cleared = before - this.state.queue.length;
|
|
if (cleared) await this.event('info', `Cleared ${cleared} completed download job${cleared === 1 ? '' : 's'} from queue`, { source: source || 'all', cleared });
|
|
return { cleared };
|
|
}
|
|
|
|
async reorder(ids) {
|
|
if (!Array.isArray(ids) || ids.length !== this.state.queue.length || new Set(ids).size !== ids.length) throw new Error('Order must contain every queue id exactly once');
|
|
const byId = new Map(this.state.queue.map(item => [item.id, item]));
|
|
if (ids.some(id => !byId.has(id))) throw new Error('Order contains an unknown queue id');
|
|
for (const current of Object.values(this.state.currentIds || {})) {
|
|
if (ids.indexOf(current) !== this.state.queue.findIndex(item => item.id === current)) throw new Error('Active items cannot be moved while downloading');
|
|
}
|
|
this.state.queue = ids.map((id, index) => ({ ...byId.get(id), order: index }));
|
|
await this.event('info', 'Queue reordered');
|
|
}
|
|
|
|
async updateSettings(patch) {
|
|
this.state.settings = validateSettings(patch, this.state.settings);
|
|
this.scheduleDiscovery();
|
|
await this.event('info', 'Scraper settings updated', { settings: this.state.settings, nextDiscoveryAt: this.state.nextDiscoveryAt });
|
|
return this.state.settings;
|
|
}
|
|
|
|
async control(action, id) {
|
|
if (action === 'pause-source' || action === 'resume-source') {
|
|
const source = String(id || '');
|
|
if (!this.adapterList.some(adapter => adapter.id === source)) throw new Error('Unknown source');
|
|
if (action === 'resume-source' && this.state.queue.some(item => item.source === source
|
|
&& item.status === 'error' && item.failure?.kind === 'access')) {
|
|
throw new Error('Review the source access block before resuming');
|
|
}
|
|
if (action === 'pause-source') {
|
|
this.state.sourcePaused[source] = true;
|
|
const activeId = this.state.currentIds[source];
|
|
const activeItem = this.state.queue.find(item => item.id === activeId);
|
|
if (activeItem) activeItem.stopReason = 'paused';
|
|
} else delete this.state.sourcePaused[source];
|
|
this.recomputeStatus();
|
|
await this.event('info', `${source} source ${action === 'pause-source' ? 'paused' : 'resumed'} by operator`, { source });
|
|
return;
|
|
}
|
|
if (action === 'inspect' || action === 'confirm') {
|
|
const item = this.state.queue.find(entry => entry.id === id);
|
|
if (!item || Object.values(this.state.currentIds || {}).includes(item.id)) throw new Error('Job not found or already active');
|
|
// Confirmation only changes the persisted queue. The work loop still
|
|
// refuses to start jobs for a paused source, so an admin can approve an
|
|
// already inspected edition without accidentally retrying blocked access.
|
|
if (item.failure?.kind === 'access' || (action === 'inspect' && this.state.sourcePaused[item.source])) {
|
|
throw new Error('Review the source access block before inspecting');
|
|
}
|
|
if (action === 'confirm') {
|
|
if (item.status !== 'ready' || item.kind !== 'inspection') throw new Error('Inspect the source first');
|
|
item.confirmedSourceTitle = item.sourceTitle;
|
|
// Prefer the inspected canonical title when metadata lookup was derived
|
|
// from the URL/title fallback. Keep an explicitly supplied search term.
|
|
const canonicalTitle = String(item.sourceTitle || '').trim();
|
|
const search = String(item.metadataSearch || '').trim();
|
|
const seriesId = String(item.seriesId || '').trim();
|
|
const fallbackSearch = !item.metadataSearchExplicit
|
|
&& (!search || search === seriesId || search === String(item.title || '').trim());
|
|
if (canonicalTitle && fallbackSearch) {
|
|
item.metadataSearch = canonicalTitle;
|
|
item.metadataSearchExplicit = true;
|
|
}
|
|
if (canonicalTitle && String(item.title || '') === String(item.seriesId || '')) item.title = canonicalTitle;
|
|
item.kind = 'download';
|
|
} else {
|
|
if (!['unavailable', 'quarantined', 'complete', 'ready'].includes(item.status)) throw new Error('Only inactive finished, unavailable or quarantined jobs can be inspected again');
|
|
item.inspectionFromComplete = !!item.completed && Number(item.publishedCount || 0) > 0;
|
|
item.kind = 'inspection'; delete item.confirmedSourceTitle;
|
|
}
|
|
item.status = 'queued'; delete item.error; delete item.failure; delete item.nextAttemptAt;
|
|
await this.event('info', action === 'confirm' ? `Download confirmed: ${item.sourceTitle}` : `Source inspection queued: ${item.title}`, { id: item.id, source: item.source });
|
|
return;
|
|
}
|
|
if (action === 'pause') {
|
|
this.state.paused = true;
|
|
this.state.status = 'pausing';
|
|
await this.event('info', 'Pause requested');
|
|
return;
|
|
}
|
|
if (action === 'resume') {
|
|
this.state.paused = false;
|
|
this.recomputeStatus();
|
|
await this.event('info', 'Queue resumed');
|
|
return;
|
|
}
|
|
if (action === 'retry') {
|
|
const item = this.state.queue.find(entry => entry.id === id) || this.state.queue.find(entry => entry.status === 'error');
|
|
if (!item) throw new Error('No failed queue item to retry');
|
|
if (Object.values(this.state.currentIds || {}).includes(item.id)) throw new Error('That item is already active');
|
|
const source = this.resolveAdapter(item.url)?.adapter.id || item.source;
|
|
const otherAccessError = this.state.queue.some(entry => entry.id !== item.id && entry.source === source && entry.status === 'error' && entry.failure?.kind === 'access');
|
|
if (otherAccessError) throw new Error('Review the access-blocked job for this source first');
|
|
// Older queued records may still carry the URL slug/ID as an automatic
|
|
// metadata query. A retry must repair that fallback too (not just confirm).
|
|
const canonicalTitle = String(item.sourceTitle || '').trim();
|
|
const search = String(item.metadataSearch || '').trim();
|
|
const seriesId = String(item.seriesId || '').trim();
|
|
if (canonicalTitle && !item.metadataSearchExplicit
|
|
&& (!search || search === seriesId || search === String(item.title || '').trim())) {
|
|
item.metadataSearch = canonicalTitle;
|
|
if (String(item.title || '') === seriesId) item.title = canonicalTitle;
|
|
}
|
|
item.status = 'queued'; delete item.error; delete item.failure; delete item.nextAttemptAt;
|
|
item.recoveryAttempts = 0;
|
|
item.stallRecoveries = 0;
|
|
delete this.state.sourcePaused[source]; delete this.state.circuits[source];
|
|
for (const incident of this.state.incidents) if (incident.jobId === item.id && incident.state === 'open') { incident.state = 'resolved'; incident.resolution = 'operator-retry'; }
|
|
this.state.lastError = null; this.state.paused = false; this.recomputeStatus();
|
|
await this.event('info', `Retrying ${item.title}`, { id: item.id, title: item.title });
|
|
return;
|
|
}
|
|
if (action === 'skip') {
|
|
const item = this.state.queue.find(entry => entry.id === id);
|
|
if (!item || this.state.currentIds[item.source] === item.id) throw new Error('Job not found or it is active');
|
|
item.status = 'skipped';
|
|
if (item.failure?.scope === 'job') delete this.state.sourcePaused[item.source];
|
|
await this.event('info', `Skipped ${item.title}`, { id: item.id, source: item.source });
|
|
return;
|
|
}
|
|
throw new Error('Unknown control action');
|
|
}
|
|
|
|
async recover(input) {
|
|
const item = this.state.queue.find(entry => entry.id === input.id);
|
|
if (!item) throw new Error('Job not found');
|
|
// Compare observations so a stale watcher cannot disturb a progressing worker.
|
|
if (Object.values(this.state.currentIds).includes(item.id)) {
|
|
const job = await this.jobState(item);
|
|
if (!job) return { changed: false };
|
|
if (progressKey(job) !== input.progressKey || this.state.paused || item.stopReason) return { changed: false };
|
|
if (Date.now() - Date.parse(job?.updated || item.started) < 900000) return { changed: false };
|
|
if ((item.stallRecoveries || 0) >= 1) {
|
|
item.stopReason = 'quarantined';
|
|
} else { item.stallRecoveries = (item.stallRecoveries || 0) + 1; item.stopReason = 'recovering'; }
|
|
await this.event('warning', `Sentinel requested recovery at the next checkpoint for ${item.title}`, { id: item.id, source: item.source });
|
|
return { changed: true };
|
|
}
|
|
return { changed: false };
|
|
}
|
|
|
|
recomputeStatus() {
|
|
const activeIds = Object.values(this.state.currentIds || {});
|
|
this.state.currentId = activeIds[0] || null;
|
|
if (this.stopping) this.state.status = 'stopping';
|
|
else if (this.state.paused) this.state.status = activeIds.length ? 'pausing' : 'paused';
|
|
else if (activeIds.length) this.state.status = 'downloading';
|
|
else if (this.state.queue.some(item => item.status === 'retry_wait')) this.state.status = 'retry_wait';
|
|
else if (this.state.queue.some(item => ['error', 'quarantined'].includes(item.status))) this.state.status = 'attention';
|
|
else this.state.status = 'idle';
|
|
}
|
|
|
|
startItem(item, found) {
|
|
const source = found.adapter.id;
|
|
item.status = 'downloading'; item.started = item.kind === 'inspection' ? new Date().toISOString() : item.started || new Date().toISOString(); delete item.error;
|
|
this.state.currentIds[source] = item.id;
|
|
this.state.lastError = null;
|
|
this.recomputeStatus();
|
|
const entry = { id: item.id, source, promise: null };
|
|
this.active.set(source, entry);
|
|
entry.promise = this.runItem(item, found, source).finally(async () => {
|
|
this.active.delete(source);
|
|
if (this.state.currentIds[source] === item.id) delete this.state.currentIds[source];
|
|
this.recomputeStatus();
|
|
await this.persist();
|
|
}).catch(error => { console.error('[scraper] persistence failed:', safeMessage(error)); this.state.paused = true; });
|
|
}
|
|
|
|
async runItem(item, found, source) {
|
|
await this.event('info', `${item.kind === 'discovery' ? 'Checking' : 'Starting'} ${item.title}`, { id: item.id, title: item.title, source, kind: item.kind });
|
|
try {
|
|
if (item.kind === 'discovery' || item.kind === 'inspection') {
|
|
await this.runDiscovery(item, found, source);
|
|
return;
|
|
}
|
|
let revisionNotifications = 0;
|
|
const result = await found.adapter.run(item, {
|
|
libraryRoot: this.libraryRoot, settings: settingsForSource(this.state.settings, source),
|
|
shouldStop: () => this.stopping ? 'stopped' : this.state.paused ? 'paused' : item.stopReason || false,
|
|
onCatalog: catalog => this.recordCatalog(item, catalog),
|
|
onPublished: async details => {
|
|
revisionNotifications++;
|
|
this.state.publishedRevision = Number(this.state.publishedRevision || 0) + 1;
|
|
const mangaId = item.libraryMangaId || (details.mangaDir && path.basename(details.mangaDir));
|
|
if (mangaId) this.recordUpdateCheck({ ...item, libraryMangaId: mangaId }, { lastDownloadedAt: new Date().toISOString() });
|
|
this.invalidateInventory();
|
|
await this.event('info', `Published ${item.title}${details.chapter ? ` chapter ${details.chapter}` : ''}`, {
|
|
id: item.id, title: item.title, source, ...(details.chapter ? { chapter: details.chapter } : {}),
|
|
});
|
|
},
|
|
});
|
|
const finalPath = typeof result === 'string' ? result : result?.finalPath || null;
|
|
if (!revisionNotifications && (typeof result === 'string' || Number(result?.publishedCount || 0) > 0)) {
|
|
this.state.publishedRevision = Number(this.state.publishedRevision || 0) + 1;
|
|
this.invalidateInventory();
|
|
}
|
|
item.status = 'complete'; item.completed = new Date().toISOString(); item.finalPath = finalPath;
|
|
delete this.state.circuits[source];
|
|
for (const incident of this.state.incidents) if (incident.jobId === item.id && incident.state === 'open') { incident.state = 'resolved'; incident.resolution = 'recovered'; }
|
|
item.publishedCount = Number(result?.publishedCount ?? 0);
|
|
if (item.publishedCount > 0 && !revisionNotifications) {
|
|
const mangaId = item.libraryMangaId || (finalPath && path.basename(finalPath));
|
|
if (mangaId) this.recordUpdateCheck({ ...item, libraryMangaId: mangaId }, { lastDownloadedAt: item.completed });
|
|
}
|
|
item.selectedCount = Number(result?.selectedCount ?? 0);
|
|
this.state.history.push({ id: item.id, title: item.title, sourceTitle: item.sourceTitle, availability: item.availability, source, status: 'complete', time: item.completed, finalPath, publishedCount: item.publishedCount, selectedCount: item.selectedCount });
|
|
await this.event('info', `Completed ${item.title}`, { id: item.id, title: item.title, source, finalPath, publishedCount: item.publishedCount });
|
|
} catch (error) {
|
|
if (error instanceof ArchiveControlError) {
|
|
item.status = error.reason === 'quarantined' ? 'quarantined' : 'queued';
|
|
delete item.stopReason;
|
|
await this.event('info', `${item.title} ${error.reason}`, { id: item.id, title: item.title, source });
|
|
return;
|
|
}
|
|
const failure = classifyFailure(error);
|
|
item.failure = failure; item.error = failure.message;
|
|
recordFailure(this.state.sourceLedger ||= {}, item, failure);
|
|
const incident = { id: crypto.randomUUID(), at: new Date().toISOString(), jobId: item.id, source, title: item.title, failure, state: 'open' };
|
|
this.state.incidents.push(incident); this.state.incidents = this.state.incidents.slice(-200);
|
|
if (failure.kind === 'unavailable') { item.status = 'unavailable'; incident.state = 'resolved'; incident.resolution = 'skipped-unavailable'; }
|
|
else if (failure.retryable && (item.recoveryAttempts || 0) < 1) {
|
|
item.recoveryAttempts = (item.recoveryAttempts || 0) + 1;
|
|
item.status = 'retry_wait'; item.nextAttemptAt = new Date(Date.now() + this.retryDelayMs).toISOString();
|
|
this.state.circuits[source] = { state: 'open', until: item.nextAttemptAt, jobId: item.id, reason: 'transient' };
|
|
} else if (failure.scope === 'source') { item.status = 'error'; this.state.sourcePaused[source] = true; }
|
|
else if (failure.scope === 'global') { item.status = 'error'; this.state.paused = true; }
|
|
else {
|
|
item.status = 'quarantined';
|
|
const similar = this.state.incidents.filter(i => i.source === source && i.failure.kind === 'adapter' && Date.now() - Date.parse(i.at) < 3600000);
|
|
if (new Set(similar.map(i => i.jobId)).size >= 2) this.state.sourcePaused[source] = true;
|
|
}
|
|
await this.event(failure.kind === 'unavailable' ? 'warning' : 'error', failure.message, { id: item.id, title: item.title, source, failure, nextAttemptAt: item.nextAttemptAt || null });
|
|
}
|
|
}
|
|
|
|
async workLoop() {
|
|
while (!this.stopping) {
|
|
this.runtimeHeartbeat = Date.now();
|
|
if (!this.state.paused) {
|
|
for (const item of this.state.queue) {
|
|
if (item.status === 'retry_wait' && Date.parse(item.nextAttemptAt) <= Date.now()) { item.status = 'queued'; delete this.state.circuits[item.source]; }
|
|
if (item.status !== 'queued') continue;
|
|
const found = this.resolveAdapter(item.url);
|
|
if (!found) {
|
|
item.status = 'quarantined'; item.error = 'Adapter no longer available';
|
|
await this.event('error', item.error, { id: item.id, title: item.title, source: item.source });
|
|
continue;
|
|
}
|
|
const source = found.adapter.id;
|
|
if (this.active.has(source) || this.state.sourcePaused[source]) continue;
|
|
if (this.state.circuits[source]?.state === 'open') continue;
|
|
if (item.libraryMangaId && [...this.active.values()].some(entry => {
|
|
const activeItem = this.state.queue.find(candidate => candidate.id === entry.id);
|
|
return activeItem?.libraryMangaId === item.libraryMangaId;
|
|
})) continue;
|
|
this.startItem(item, found);
|
|
}
|
|
}
|
|
const prior = this.state.status;
|
|
this.recomputeStatus();
|
|
if (prior !== this.state.status) await this.persist();
|
|
await sleep(500);
|
|
}
|
|
}
|
|
|
|
async close() {
|
|
if (!this.state) return;
|
|
this.stopping = true;
|
|
if (this.discoveryTimer) clearTimeout(this.discoveryTimer);
|
|
this.discoveryTimer = null;
|
|
this.recomputeStatus();
|
|
await this.persist();
|
|
const timeout = new Promise(resolve => { const timer = setTimeout(resolve, 30000); timer.unref(); });
|
|
const workers = [...this.active.values()].map(entry => entry.promise).filter(Boolean);
|
|
const drained = await Promise.race([Promise.allSettled([this.loopPromise || Promise.resolve(), ...workers]).then(() => true), timeout.then(() => false)]);
|
|
await this.writeChain;
|
|
// Keep the lock until the process exits if an in-flight worker did not drain.
|
|
if (drained) await fs.rm(this.lockFile, { force: true });
|
|
}
|
|
}
|
|
|
|
module.exports = { ScraperService, DEFAULT_SETTINGS, validateSettings, settingsForSource };
|