'use strict'; const crypto = require('node:crypto'); const { AniListStop, AniListSkip } = require('./anilist'); const { topTagsFor } = require('./anchors'); const BUDGETS = { nightly: 300, admin: 300, user: 60 }; const REFRESH_COOLDOWN_MS = 3600000; const MAX_CONSECUTIVE_FAILURES = 5; const CREATOR_MIN_WEIGHT = 3; // Thrown out of a run to unwind every stage at once. class StopRun extends Error { constructor(status, stopReason) { super(stopReason); this.status = status; this.stopReason = stopReason; } } async function runHarvest({ kind, targets, cache, client, budget }) { let requests = 0, failures = 0; const seen = new Set(); // One request: skipped when already done this run or still fresh in the cache, otherwise awaited in turn. const fetchInto = async (cacheKind, key, call) => { const id = `${cacheKind}\n${key}`; if (seen.has(id)) return; seen.add(id); if (cache.isFresh(cacheKind, key)) return; if (requests >= budget) throw new StopRun('success', 'budget'); requests++; let data; try { data = await call(); } catch (error) { if (error instanceof AniListStop) throw new StopRun('stopped', error.reason); if (!(error instanceof AniListSkip)) throw error; if (++failures >= MAX_CONSECUTIVE_FAILURES) throw new StopRun('stopped', 'upstream-errors'); return; } failures = 0; // A null title is not a manga format: nothing worth keeping. if (data !== null) cache.put(cacheKind, key, data); }; const anchors = targets.flatMap(target => target.anchors); const recordOf = id => cache.get('title', id)?.data || null; try { for (const { anilistId } of anchors) await fetchInto('title', anilistId, () => client.title(anilistId)); for (const { anilistId } of anchors) await fetchInto('recs', anilistId, () => client.recs(anilistId)); for (const anchor of anchors) { if (anchor.weight < CREATOR_MIN_WEIGHT) continue; for (const { staffId } of recordOf(anchor.anilistId)?.authors || []) await fetchInto('creator', staffId, () => client.creator(staffId)); } for (const target of targets) { for (const tag of topTagsFor(target.anchors, recordOf, { mature: target.mature })) await fetchInto('tagTop', `${tag.kind}:${tag.name}`, () => client.tagTop(tag)); } // With nobody to harvest for, even the global list is not worth a request. if (targets.length) await fetchInto('fresh', 'global', () => client.fresh()); } catch (error) { // The caller records how many requests a local fault had already spent. if (!(error instanceof StopRun)) throw Object.assign(error, { requests }); return { requests, budget, status: error.status, stopReason: error.stopReason }; } return { requests, budget, status: 'success', stopReason: null }; } const localDate = date => [date.getFullYear(), String(date.getMonth() + 1).padStart(2, '0'), String(date.getDate()).padStart(2, '0')].join('-'); function shouldRunNightly({ now, harvestAt, lastNightlyDate, enabled }) { if (!enabled) return false; const match = /^(\d{2}):(\d{2})$/.exec(harvestAt || ''); if (!match) return false; if (now.getHours() * 60 + now.getMinutes() < Number(match[1]) * 60 + Number(match[2])) return false; return localDate(now) !== lastNightlyDate; } const httpError = (status, message, extra = {}) => Object.assign(new Error(message), { status }, extra); // These reasons mean AniList pushed back: no further calls until a nightly or admin run. const BLOCKING_REASONS = ['rate-limited', 'forbidden', 'challenge']; const blockingReason = record => record?.status === 'stopped' && BLOCKING_REASONS.includes(record.stopReason) ? record.stopReason : null; class Harvester { constructor({ cache, client, store, targetsFor, onDone, now = Date.now }) { Object.assign(this, { cache, client, store, targetsFor, onDone, now }); this.queue = []; this.active = null; this.lastAccepted = new Map(); // The pause outlives a restart through the persisted harvest record. try { this.blockedReason = blockingReason(store.load?.()?.harvest); } catch { this.blockedReason = null; } } nextAllowedAt(user) { const last = this.lastAccepted.get(user); if (last === undefined || this.now() - last >= REFRESH_COOLDOWN_MS) return null; return new Date(last + REFRESH_COOLDOWN_MS).toISOString(); } jobs() { return [this.active, ...this.queue].filter(Boolean); } request({ kind, user }) { if (!Object.hasOwn(BUDGETS, kind)) throw httpError(400, `Unknown harvest kind: ${kind}`); if (kind === 'user') { if (!user) throw httpError(400, 'A user refresh needs a user'); if (this.jobs().some(job => job.kind === 'user' && job.user === user)) throw httpError(409, 'A refresh for this user is already queued or running'); if (this.blockedReason) throw httpError(503, `AniList paused the harvest: ${this.blockedReason}`, { blockedReason: this.blockedReason }); const nextAllowedAt = this.nextAllowedAt(user); if (nextAllowedAt) throw httpError(429, 'Refresh is cooling down', { nextAllowedAt }); this.lastAccepted.set(user, this.now()); } else if (this.jobs().length) { throw httpError(409, 'A harvest is already queued or running'); } else { // An accepted nightly or admin run is the one allowed retry after a pause. this.blockedReason = null; } const job = { id: crypto.randomBytes(8).toString('hex'), kind, ...(kind === 'user' ? { user } : {}) }; const position = this.jobs().length; this.queue.push(job); if (!this.active) this.drain(); return { harvestId: job.id, position }; } state(userName) { const nextAllowedAt = this.nextAllowedAt(userName); const extra = { ...(nextAllowedAt ? { nextAllowedAt } : {}), ...(this.blockedReason ? { blockedReason: this.blockedReason } : {}) }; const index = this.jobs().findIndex(job => job.kind === 'user' && job.user === userName); if (index === 0 && this.active) return { state: 'running', ...extra }; if (index >= 0) return { state: 'queued', position: index, ...extra }; return { state: 'idle', ...extra }; } current() { if (!this.active) return null; const { id, kind, user, startedAt } = this.active; return { id, kind, ...(user ? { user } : {}), startedAt }; } // Queued refreshes are dropped unrun; their users get their cooldown back for after the pause. pause(reason) { this.blockedReason = reason; for (const job of this.queue) if (job.kind === 'user') this.lastAccepted.delete(job.user); this.queue = this.queue.filter(job => job.kind !== 'user'); } // Runs synchronously up to the first await, so a job is active before request() returns. async drain() { while (this.queue.length) { const job = this.queue.shift(); job.startedAt = new Date(this.now()).toISOString(); this.active = job; try { await this.run(job); } catch (error) { console.error(`Discovery harvest ${job.id} failed:`, error); } finally { this.active = null; } } } async run(job) { const budget = BUDGETS[job.kind]; const base = { id: job.id, kind: job.kind, user: job.user ?? null }; await this.store.patch({ harvest: { ...base, status: 'running', startedAt: job.startedAt, requests: 0, budget } }); let result; try { const targets = await this.targetsFor(job.kind, job.user); result = await runHarvest({ kind: job.kind, targets, cache: this.cache, client: this.client, budget }); } catch (error) { // A local fault (cache write, bad target) must not leave the record "running" until restart. console.error(`Discovery harvest ${job.id} failed:`, error); result = { requests: error?.requests ?? 0, budget, status: 'stopped', stopReason: 'error' }; } const blocked = blockingReason(result); if (blocked) this.pause(blocked); await this.store.patch({ harvest: { ...base, status: result.status, startedAt: job.startedAt, finishedAt: new Date(this.now()).toISOString(), requests: result.requests, budget, stopReason: result.stopReason } }); try { await this.onDone(job, result); } catch (error) { console.error(`Discovery harvest ${job.id} follow-up failed:`, error); } } } module.exports = { BUDGETS, REFRESH_COOLDOWN_MS, MAX_CONSECUTIVE_FAILURES, runHarvest, shouldRunNightly, Harvester };