Single-flight discovery harvester with budgets and stop reasons (BL-0051)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

discovery/harvest.js

@@ -0,0 +1,152 @@
+'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) {
+    if (!(error instanceof StopRun)) throw error;
+    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);
+
+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();
+  }
+  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 (!(kind in BUDGETS)) 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');
+      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');
+    }
+    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 cooldown = nextAllowedAt ? { nextAllowedAt } : {};
+    const index = this.jobs().findIndex(job => job.kind === 'user' && job.user === userName);
+    if (index === 0 && this.active) return { state: 'running', ...cooldown };
+    if (index >= 0) return { state: 'queued', position: index, ...cooldown };
+    return { state: 'idle', ...cooldown };
+  }
+  current() {
+    if (!this.active) return null;
+    const { id, kind, user, startedAt } = this.active;
+    return { id, kind, ...(user ? { user } : {}), startedAt };
+  }
+  // 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 = 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: 0, budget, status: 'stopped', stopReason: 'error' };
+    }
+    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 };