Discovery harvest routes, nightly schedule and per-user refresh (BL-0051)

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

server.js

@@ -23,7 +23,11 @@ const { secret } = require('./runtime/config');
 const { PeeringState, requirePeer } = require('./peering/state');
 const peerClient = require('./peering/client');
 const { DiscoveryStore } = require('./discovery/store');
-const { generate: generateDiscovery } = require('./discovery/engine');
+const { generateAsync: generateDiscovery, mergeUserFeed } = require('./discovery/engine');
+const { DiscoveryCache } = require('./discovery/cache');
+const { createClient: createAniListClient } = require('./discovery/anilist');
+const { anchorsFor } = require('./discovery/anchors');
+const { shouldRunNightly, Harvester } = require('./discovery/harvest');
 const {
   applyPulse, applyVisit, chapterInsight, earliestCompletion, eligiblePages, emptyRecord, libraryInsight, mangaInsight, publicSummary,
 } = require('./reading-analytics');
@@ -44,6 +48,9 @@ const PEER_DATA = path.resolve(process.env.PEER_DATA || path.join(DATA, 'peering
 const peerStore = new PeeringState(PEER_DATA);
 const discoveryStore = new DiscoveryStore(DATA);
 const DISCOVERY_CACHE = path.join(DATA, 'scraper', 'metadata', 'anilist');
+const DISCOVERY_HARVEST_CACHE = path.join(DATA, 'discovery-cache');
+const HARVEST_TIME = /^([01]\d|2[0-3]):[0-5]\d$/;
+const DISCOVERY_HARVEST_AT = HARVEST_TIME.test(process.env.DISCOVERY_HARVEST_AT || '') ? process.env.DISCOVERY_HARVEST_AT : '03:30';
 let discoveryRun = null, discoveryError = null;
 const PUB = path.join(__dirname, 'public');
 const IMG = /\.(jpe?g|png|webp|gif|avif|bmp)$/i;
@@ -78,8 +85,9 @@ const readingAccess = createReadingAccess({ schedule: readingSchedule,
   getProgress: (user, mangaId) => db.progress[user]?.[mangaId] || {},
   getMangaTitle: mangaId => lib.byId.get(mangaId)?.title || null });
 
-class HttpError extends Error { constructor(code, msg) { super(msg); this.code = code; } }
-function bad(code, msg) { throw new HttpError(code, msg); }
+// `extra` fields are sent next to `error`, e.g. `nextAllowedAt` on a 429.
+class HttpError extends Error { constructor(code, msg, extra = {}) { super(msg); this.code = code; this.extra = extra; } }
+function bad(code, msg, extra) { throw new HttpError(code, msg, extra); }
 function peerForRoute(id) {
   try { return requirePeer(peerStore.load(), id); }
   catch { bad(404, 'Peer not found'); }
@@ -241,6 +249,26 @@ function readDiscovery() {
   try { return discoveryStore.load(); }
   catch (error) { bad(503, error.message); }
 }
+// Scoring is local: it reads the harvest cache and never calls AniList. With `onlyUser` only that
+// feed is rebuilt and the other users' feeds are carried over into the same publish.
+async function rescoreNow(onlyUser) {
+  const startedAt = new Date().toISOString();
+  let queuedTitles = [], queueAvailable = true;
+  try { queuedTitles = (await scraperRequest('/api/state')).queue?.filter(item => !['complete', 'removed'].includes(item.status)).map(item => item.title) || []; }
+  catch { queueAvailable = false; }
+  const result = await generateDiscovery({ db, library: lib.mangas, greatest: greatestCatalog,
+    cacheDir: DISCOVERY_CACHE, harvestCache, queuedTitles, onlyUser });
+  // Read the prior store only now, right before the synchronous publish, so no other publish slips between.
+  const value = onlyUser === undefined ? result : { ...result, ...mergeUserFeed(discoveryStore.load(), result, onlyUser) };
+  return { published: discoveryStore.publish({ ...value, startedAt }), queueAvailable };
+}
+// One rescore at a time: an admin run and a harvest follow-up must not publish over each other.
+let rescoring = Promise.resolve();
+function rescore({ onlyUser } = {}) {
+  const run = rescoring.then(() => rescoreNow(onlyUser));
+  rescoring = run.catch(() => {});
+  return run;
+}
 function startDiscoveryRun() {
   if (discoveryRun?.status === 'running') bad(409, 'A discovery run is already active');
   readDiscovery(); // Refuse a new run if the on-disk store is corrupt.
@@ -249,12 +277,7 @@ function startDiscoveryRun() {
   discoveryError = null;
   setImmediate(async () => {
     try {
-      let queuedTitles = [], queueAvailable = true;
-      try { queuedTitles = (await scraperRequest('/api/state')).queue?.filter(item => !['complete', 'removed'].includes(item.status)).map(item => item.title) || []; }
-      catch { queueAvailable = false; }
-      const result = generateDiscovery({ db, library: lib.mangas, greatest: greatestCatalog,
-        cacheDir: DISCOVERY_CACHE, queuedTitles });
-      const published = discoveryStore.publish({ ...result, startedAt });
+      const { published, queueAvailable } = await rescore();
       discoveryRun = { id, status: 'success', startedAt, finishedAt: published.lastRun.finishedAt,
         candidates: published.lastRun.candidateCount, queueAvailable };
     } catch (error) {
@@ -266,6 +289,56 @@ function startDiscoveryRun() {
   return discoveryRun;
 }
 
+// ---------- discovery harvest (AniList seeds, BL-0051) ----------
+const harvestCache = new DiscoveryCache(DISCOVERY_HARVEST_CACHE);
+const discoverySettings = state => ({ harvestEnabled: state.settings?.harvestEnabled ?? true,
+  harvestAt: state.settings?.harvestAt ?? DISCOVERY_HARVEST_AT });
+const localDate = date => [date.getFullYear(), String(date.getMonth() + 1).padStart(2, '0'), String(date.getDate()).padStart(2, '0')].join('-');
+function nextNightlyAt(state, now = new Date()) {
+  const { harvestEnabled, harvestAt } = discoverySettings(state);
+  if (!harvestEnabled) return null;
+  const [hours, minutes] = harvestAt.split(':').map(Number);
+  const slot = new Date(now);
+  slot.setHours(hours, minutes, 0, 0);
+  if (state.nightly?.lastDate === localDate(now)) slot.setDate(slot.getDate() + 1);
+  else if (slot < now) return now.toISOString(); // due: the next minute's check starts it
+  return slot.toISOString();
+}
+function harvestTargets(kind, userName) {
+  const users = kind === 'user' ? [db.users[userName]].filter(Boolean) : Object.values(db.users);
+  return users.filter(user => settingsOf(user).discoveryEnabled).map(user => ({ user,
+    anchors: anchorsFor({ user, db, library: lib.mangas, greatest: greatestCatalog }), mature: !!settingsOf(user).discoveryMature }));
+}
+// Queued jobs live in memory only; a run cut off by a restart is recorded as interrupted.
+try {
+  const { harvest } = discoveryStore.load();
+  if (harvest?.status === 'running') discoveryStore.patch({ harvest: { ...harvest, status: 'interrupted', finishedAt: new Date().toISOString() } });
+} catch (error) { console.error('[discovery]', error.message); }
+const harvester = new Harvester({ cache: harvestCache, client: createAniListClient(), store: discoveryStore,
+  targetsFor: harvestTargets, onDone: job => rescore(job.kind === 'user' ? { onlyUser: job.user } : {}) });
+function requestHarvest(args) {
+  readDiscovery(); // A corrupt store fails closed before any AniList call.
+  try { return harvester.request(args); }
+  catch (error) {
+    if (!error.status) throw error;
+    const extra = {};
+    for (const key of ['nextAllowedAt', 'blockedReason']) if (error[key]) extra[key] = error[key];
+    bad(error.status, error.message, extra);
+  }
+}
+function checkNightly() {
+  let state;
+  try { state = discoveryStore.load(); } catch { return; }
+  const now = new Date(), { harvestEnabled, harvestAt } = discoverySettings(state);
+  if (!shouldRunNightly({ now, harvestAt, lastNightlyDate: state.nightly?.lastDate, enabled: harvestEnabled })) return;
+  if (harvester.current()) return;
+  // The date is written only once the run was accepted; a 409 (refresh in flight) retries next minute.
+  try { harvester.request({ kind: 'nightly' }); }
+  catch (error) { if (error.status !== 409) console.error('[discovery] nightly harvest:', error.message); return; }
+  try { discoveryStore.patch({ nightly: { lastDate: localDate(now) } }); }
+  catch (error) { console.error('[discovery]', error.message); }
+}
+
 const mangaDexSearchCache = new Map();
 let lastMangaDexSearch = 0;
 async function searchGreatestMangaDex(entry) {
@@ -494,6 +567,7 @@ const routes = {
       discoveryReadingActivity: !!settingsOf(u).discoveryReadingActivity,
       discoveryMature: !!settingsOf(u).discoveryMature },
     generatedAt: feed?.generatedAt || null, revision: state.revision, lastRun: state.lastRun,
+    refresh: harvester.state(u.name),
     feedback: db.discoveryFeedback?.[u.name] || {},
     items: (feed?.items || []).map(item => ({ ...item, candidate: state.candidates[item.candidateId] || null })).filter(item => item.candidate) };
   },
@@ -507,14 +581,30 @@ const routes = {
     save();
     return { candidateId: id, value };
   },
+  'POST /api/discovery/refresh': ({ u }) => {
+    if (!settingsOf(u).discoveryEnabled) bad(403, 'Enable personalized discovery first');
+    return requestHarvest({ kind: 'user', user: u.name });
+  },
   'GET /api/admin/discovery': ({ u }) => {
     admin(u);
     const state = readDiscovery();
     return { revision: state.revision, lastRun: state.lastRun, currentRun: discoveryRun,
       lastError: discoveryError, optedInUsers: Object.values(db.users).filter(user => user.settings?.discoveryEnabled).length,
-      candidateCount: Object.keys(state.candidates).length };
+      candidateCount: Object.keys(state.candidates).length,
+      harvest: state.harvest, currentHarvest: harvester.current(), cache: harvestCache.stats(),
+      settings: discoverySettings(state), nextNightlyAt: nextNightlyAt(state) };
   },
   'POST /api/admin/discovery/run': ({ u }) => { admin(u); return startDiscoveryRun(); },
+  'POST /api/admin/discovery/harvest': ({ u }) => { admin(u); return requestHarvest({ kind: 'admin' }); },
+  'POST /api/admin/discovery/settings': ({ u, b }) => {
+    admin(u);
+    const keys = b && typeof b === 'object' && !Array.isArray(b) ? Object.keys(b).sort().join() : '';
+    if (keys !== 'harvestAt,harvestEnabled' || typeof b.harvestEnabled !== 'boolean' || typeof b.harvestAt !== 'string' || !HARVEST_TIME.test(b.harvestAt))
+      bad(400, 'Send exactly { harvestEnabled: true|false, harvestAt: "HH:MM" }');
+    readDiscovery();
+    const state = discoveryStore.patch({ settings: { harvestEnabled: b.harvestEnabled, harvestAt: b.harvestAt } });
+    return { settings: discoverySettings(state), nextNightlyAt: nextNightlyAt(state) };
+  },
 
   'GET /api/greatest': ({ u }) => ({
     title: greatestCatalog.title, updatedAt: greatestCatalog.updatedAt, periods: greatestCatalog.periods,
@@ -971,7 +1061,7 @@ http.createServer(async (req, res) => {
   } catch (e) {
     const expected = e instanceof HttpError || e instanceof CatalogError;
     if (!expected) console.error(e);
-    if (!res.headersSent) sendJson(res, expected ? e.code : 500, { error: expected ? e.message : 'Internal error' });
+    if (!res.headersSent) sendJson(res, expected ? e.code : 500, expected ? { error: e.message, ...e.extra } : { error: 'Internal error' });
   }
 }).listen(PORT, () => console.log(`MangaCircus on http://localhost:${PORT}  (library: ${LIB}, data: ${DATA})`));
 
@@ -1010,3 +1100,7 @@ async function pollScraper() {
 }
 pollScraper();
 setInterval(pollScraper, 15000).unref();
+
+// Nightly discovery harvest; the delayed first check also covers a slot missed while the reader was down.
+setInterval(checkNightly, 60000).unref();
+setTimeout(checkNightly, 30000).unref();