MangaCircus/test/discovery-harvest.test.js

'use strict';

const test = require('node:test');
const assert = require('node:assert/strict');
const fs = require('node:fs');
const os = require('node:os');
const path = require('node:path');
const { DiscoveryCache } = require('../discovery/cache');
const { AniListStop, AniListSkip } = require('../discovery/anilist');
const {
  BUDGETS, REFRESH_COOLDOWN_MS, MAX_CONSECUTIVE_FAILURES, runHarvest, shouldRunNightly, Harvester,
} = require('../discovery/harvest');

const tempCache = (t) => {
  const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'discovery-harvest-'));
  t.after(() => fs.rmSync(dir, { recursive: true, force: true }));
  return new DiscoveryCache(dir);
};

const record = (id, { tags = [], authors = [] } = {}) => ({ anilistId: id, title: `T${id}`, tags, authors });

// Fake client: answers from fixture maps, logs every call, and throws scripted errors by call number.
function fakeClient({ titles = {}, errors = {} } = {}) {
  const calls = [];
  const answer = (name, key, value) => {
    calls.push(`${name}:${key}`);
    const error = errors[calls.length];
    if (error) throw error;
    return value;
  };
  return {
    calls,
    async title(id) { return answer('title', id, id in titles ? titles[id] : record(id)); },
    async recs(id) { return answer('recs', id, [{ anilistId: id * 100 }]); },
    async creator(staffId) { return answer('creator', staffId, [{ anilistId: staffId * 10 }]); },
    async tagTop({ name, kind }) { return answer('tagTop', `${kind}:${name}`, [{ anilistId: 1 }]); },
    async fresh() { return answer('fresh', 'global', [{ anilistId: 2 }]); },
  };
}

const anchor = (anilistId, weight = 1) => ({ anilistId, weight, source: 'owned', title: null });

test('constants', () => {
  assert.deepEqual(BUDGETS, { nightly: 300, admin: 300, user: 60 });
  assert.equal(REFRESH_COOLDOWN_MS, 3600000);
  assert.equal(MAX_CONSECUTIVE_FAILURES, 5);
});

test('plan order and dedupe', async (t) => {
  const cache = tempCache(t);
  const titles = {
    1: record(1, { tags: [{ name: 'Drama', kind: 'genre', rank: 100, isAdult: false }], authors: [{ staffId: 9, name: 'A', role: 'Story' }] }),
    2: record(2, { tags: [{ name: 'Drama', kind: 'genre', rank: 100, isAdult: false }] }),
    3: record(3, { tags: [{ name: 'Gore', kind: 'tag', rank: 80, isAdult: false }], authors: [{ staffId: 8, name: 'B', role: 'Art' }] }),
  };
  const client = fakeClient({ titles });
  const targets = [
    { user: 'ann', anchors: [anchor(1, 5), anchor(2, 1)], mature: false },
    { user: 'bob', anchors: [anchor(1, 3), anchor(3, 1)], mature: false },
  ];
  const result = await runHarvest({ kind: 'admin', targets, cache, client, budget: 300 });
  assert.deepEqual(client.calls, [
    'title:1', 'title:2', 'title:3',
    'recs:1', 'recs:2', 'recs:3',
    'creator:9',
    'tagTop:genre:Drama', 'tagTop:tag:Gore',
    'fresh:global',
  ]);
  assert.deepEqual(result, { requests: 10, budget: 300, status: 'success', stopReason: null });
  assert.equal(cache.get('title', 1).data.anilistId, 1);
  assert.deepEqual(cache.get('recs', 2).data, [{ anilistId: 200 }]);
  assert.deepEqual(cache.get('creator', 9).data, [{ anilistId: 90 }]);
  assert.deepEqual(cache.get('tagTop', 'genre:Drama').data, [{ anilistId: 1 }]);
  assert.deepEqual(cache.get('fresh', 'global').data, [{ anilistId: 2 }]);
});

test('non-manga titles are counted but not cached', async (t) => {
  const cache = tempCache(t);
  const client = fakeClient({ titles: { 4: null } });
  const result = await runHarvest({ kind: 'user', targets: [{ user: 'ann', anchors: [anchor(4, 5)], mature: false }], cache, client, budget: 60 });
  assert.equal(cache.get('title', 4), null);
  assert.deepEqual(client.calls, ['title:4', 'recs:4', 'fresh:global']);
  assert.equal(result.requests, 3);
});

test('fresh cache entries are skipped', async (t) => {
  const cache = tempCache(t);
  cache.put('title', 1, record(1));
  cache.put('recs', 1, []);
  cache.put('fresh', 'global', []);
  const client = fakeClient();
  const result = await runHarvest({ kind: 'user', targets: [{ user: 'ann', anchors: [anchor(1), anchor(2)], mature: false }], cache, client, budget: 60 });
  assert.deepEqual(client.calls, ['title:2', 'recs:2']);
  assert.equal(result.requests, 2);
  assert.equal(result.stopReason, null);
});

test('budget stops with reason budget', async (t) => {
  const cache = tempCache(t);
  const client = fakeClient();
  const targets = [{ user: 'ann', anchors: [anchor(1), anchor(2), anchor(3)], mature: false }];
  const result = await runHarvest({ kind: 'user', targets, cache, client, budget: 3 });
  assert.deepEqual(result, { requests: 3, budget: 3, status: 'success', stopReason: 'budget' });
  assert.equal(client.calls.length, 3);
});

test('429 stops at once', async (t) => {
  const cache = tempCache(t);
  const client = fakeClient({ errors: { 2: new AniListStop('rate-limited') } });
  const targets = [{ user: 'ann', anchors: [anchor(1), anchor(2), anchor(3)], mature: false }];
  const result = await runHarvest({ kind: 'admin', targets, cache, client, budget: 300 });
  assert.deepEqual(result, { requests: 2, budget: 300, status: 'stopped', stopReason: 'rate-limited' });
  assert.equal(client.calls.length, 2);
  assert.equal(cache.get('title', 2), null);
});

test('network error stops with upstream-errors after 5', async (t) => {
  const skip = () => new AniListSkip('timeout');
  const targets = [{ user: 'ann', anchors: [1, 2, 3, 4, 5, 6, 7, 8].map((id) => anchor(id)), mature: false }];

  const client = fakeClient({ errors: { 1: skip(), 2: skip(), 3: skip(), 4: skip(), 5: skip() } });
  const result = await runHarvest({ kind: 'admin', targets, cache: tempCache(t), client, budget: 300 });
  assert.deepEqual(result, { requests: 5, budget: 300, status: 'stopped', stopReason: 'upstream-errors' });
  assert.equal(client.calls.length, 5);

  // A success in the middle resets the counter, so the run gets past five failures in total.
  const mixed = fakeClient({ errors: { 1: skip(), 2: skip(), 3: skip(), 4: skip(), 6: skip(), 7: skip(), 8: skip() } });
  const reset = await runHarvest({ kind: 'admin', targets, cache: tempCache(t), client: mixed, budget: 300 });
  assert.equal(reset.status, 'success');
  assert.equal(reset.stopReason, null);
  assert.ok(mixed.calls.length > 8);
});

test('plan for anchorless user only fetches fresh', async (t) => {
  const client = fakeClient();
  const result = await runHarvest({ kind: 'user', targets: [{ user: 'ann', anchors: [], mature: false }], cache: tempCache(t), client, budget: 60 });
  assert.deepEqual(client.calls, ['fresh:global']);
  assert.deepEqual(result, { requests: 1, budget: 60, status: 'success', stopReason: null });
});

test('shouldRunNightly', () => {
  const at = (h, m) => new Date(2026, 9, 25, h, m);
  assert.equal(shouldRunNightly({ now: at(3, 29), harvestAt: '03:30', lastNightlyDate: null, enabled: true }), false);
  assert.equal(shouldRunNightly({ now: at(3, 30), harvestAt: '03:30', lastNightlyDate: null, enabled: true }), true);
  assert.equal(shouldRunNightly({ now: at(23, 0), harvestAt: '03:30', lastNightlyDate: '2026-10-24', enabled: true }), true);
  assert.equal(shouldRunNightly({ now: at(3, 30), harvestAt: '03:30', lastNightlyDate: '2026-10-25', enabled: true }), false);
  assert.equal(shouldRunNightly({ now: at(3, 30), harvestAt: '03:30', lastNightlyDate: null, enabled: false }), false);
});

// A run that waits until the test releases it, so queue state can be observed.
function gatedHarvester({ onDone = async () => {}, now, load }) {
  const gates = [];
  const patches = [];
  const done = [];
  const client = {
    calls: [],
    async fresh() {
      this.calls.push('fresh');
      await new Promise((resolve, reject) => gates.push({ resolve, reject }));
      return [];
    },
  };
  // Nothing stays cached, so every job makes its one fresh call and waits at the gate.
  const cache = { isFresh: () => false, get: () => null, put() {} };
  const harvester = new Harvester({
    cache,
    client,
    store: { patch(value) { patches.push(structuredClone(value)); }, ...(load ? { load } : {}) },
    targetsFor: (kind, user) => [{ user: user || 'all', anchors: [], mature: false }],
    onDone: async (job, result) => { done.push({ job, result }); await onDone(job, result); },
    now,
  });
  // Lets the waiting request finish, or fail with `error`.
  const release = async (error) => {
    for (let i = 0; !gates.length; i++) {
      if (i > 1000) throw new Error('no harvest request is waiting');
      await new Promise(setImmediate);
    }
    const gate = gates.shift();
    if (error) gate.reject(error); else gate.resolve();
    await new Promise(setImmediate);
  };
  return { harvester, patches, done, release, client };
}

const settle = async () => { for (let i = 0; i < 20; i++) await new Promise(setImmediate); };

test('harvester single-flight, positions, 409 and 429 cooldown', async (t) => {
  let clock = Date.parse('2026-10-25T10:00:00Z');
  const h = gatedHarvester({ now: () => clock });
  const { harvester } = h;

  assert.deepEqual(harvester.state('ann'), { state: 'idle' });
  assert.equal(harvester.current(), null);

  const first = harvester.request({ kind: 'user', user: 'ann' });
  assert.equal(first.position, 0);
  assert.equal(typeof first.harvestId, 'string');
  assert.equal(harvester.state('ann').state, 'running');
  assert.deepEqual(harvester.current(), { id: first.harvestId, kind: 'user', user: 'ann', startedAt: '2026-10-25T10:00:00.000Z' });

  assert.throws(() => harvester.request({ kind: 'admin' }), (error) => error.status === 409);
  assert.throws(() => harvester.request({ kind: 'nightly' }), (error) => error.status === 409);
  assert.throws(() => harvester.request({ kind: 'user', user: 'ann' }), (error) => error.status === 409);

  const second = harvester.request({ kind: 'user', user: 'bob' });
  assert.equal(second.position, 1);
  assert.deepEqual(harvester.state('bob'), { state: 'queued', position: 1, nextAllowedAt: '2026-10-25T11:00:00.000Z' });
  assert.throws(() => harvester.request({ kind: 'user', user: 'bob' }), (error) => error.status === 409);

  await h.release();
  await settle();
  assert.equal(harvester.state('bob').state, 'running');
  assert.equal(harvester.state('ann').state, 'idle');

  clock += 30 * 60000;
  assert.throws(() => harvester.request({ kind: 'user', user: 'ann' }), (error) => {
    assert.equal(error.status, 429);
    assert.equal(error.nextAllowedAt, '2026-10-25T11:00:00.000Z');
    return true;
  });
  assert.deepEqual(harvester.state('ann'), { state: 'idle', nextAllowedAt: '2026-10-25T11:00:00.000Z' });

  await h.release();
  await settle();
  assert.equal(harvester.current(), null);

  // The persisted records: running first, then the final one, for each job in order.
  assert.deepEqual(h.patches.map((p) => [p.harvest.user, p.harvest.status]),
    [['ann', 'running'], ['ann', 'success'], ['bob', 'running'], ['bob', 'success']]);
  assert.deepEqual(h.patches[0].harvest, { id: first.harvestId, kind: 'user', user: 'ann', status: 'running',
    startedAt: '2026-10-25T10:00:00.000Z', requests: 0, budget: 60 });
  assert.equal(h.patches[1].harvest.requests, 1);
  assert.equal(h.patches[1].harvest.stopReason, null);
  assert.equal(typeof h.patches[1].harvest.finishedAt, 'string');
  assert.deepEqual(h.done.map((d) => d.job.user), ['ann', 'bob']);

  clock += 30 * 60000;
  const admin = harvester.request({ kind: 'admin' });
  assert.equal(admin.position, 0);
  const ann = harvester.request({ kind: 'user', user: 'ann' });
  assert.equal(ann.position, 1);
  await h.release();
  await h.release();
  await settle();
  assert.equal(h.patches.at(-4).harvest.budget, 300);
});

test('onDone failure does not stall the queue', async (t) => {
  const errors = [];
  const original = console.error;
  console.error = (...args) => errors.push(args);
  t.after(() => { console.error = original; });
  const h = gatedHarvester({ now: () => Date.parse('2026-10-25T10:00:00Z'), onDone: async () => { throw new Error('score failed'); } });
  h.harvester.request({ kind: 'nightly' });
  h.harvester.request({ kind: 'user', user: 'ann' });
  await h.release();
  await h.release();
  await settle();
  assert.deepEqual(h.done.map((d) => d.job.kind), ['nightly', 'user']);
  assert.equal(h.harvester.current(), null);
  assert.ok(errors.length >= 2);
});

test('AniList stop pauses user refreshes until an admin or nightly run', async () => {
  let clock = Date.parse('2026-10-25T10:00:00Z');
  const h = gatedHarvester({ now: () => clock });
  const { harvester, client } = h;
  harvester.request({ kind: 'user', user: 'ann' });
  harvester.request({ kind: 'user', user: 'bob' });
  assert.equal(harvester.state('bob').state, 'queued');

  // (a) 429 while ann runs: bob's queued refresh is dropped without touching AniList.
  await h.release(new AniListStop('rate-limited'));
  await settle();
  assert.equal(client.calls.length, 1);
  assert.equal(harvester.current(), null);
  assert.deepEqual(harvester.state('bob'), { state: 'idle', blockedReason: 'rate-limited' });
  assert.deepEqual(h.patches.map((p) => [p.harvest.user, p.harvest.status, p.harvest.stopReason ?? null]),
    [['ann', 'running', null], ['ann', 'stopped', 'rate-limited']]);
  assert.deepEqual(h.done.map((d) => d.job.user), ['ann']);

  // (b) a new refresh is refused while paused, even for a user outside any cooldown.
  clock += 2 * REFRESH_COOLDOWN_MS;
  assert.throws(() => harvester.request({ kind: 'user', user: 'cid' }), (error) => {
    assert.equal(error.status, 503);
    assert.equal(error.blockedReason, 'rate-limited');
    assert.match(error.message, /AniList paused the harvest: rate-limited/);
    return true;
  });
  assert.equal(client.calls.length, 1);

  // (c) an accepted admin run is the allowed retry and lifts the pause.
  harvester.request({ kind: 'admin' });
  assert.equal(harvester.state('cid').blockedReason, undefined);
  await h.release();
  await settle();
  assert.equal(client.calls.length, 2);
  assert.equal(harvester.request({ kind: 'user', user: 'cid' }).position, 0);
  await h.release();
  await settle();
});

test('pause survives a restart through the stored harvest record', () => {
  const load = () => ({ harvest: { status: 'stopped', stopReason: 'forbidden' } });
  const { harvester } = gatedHarvester({ now: () => 0, load });
  assert.throws(() => harvester.request({ kind: 'user', user: 'ann' }),
    (error) => error.status === 503 && error.blockedReason === 'forbidden');
  assert.equal(harvester.state('ann').blockedReason, 'forbidden');

  const budgetStop = gatedHarvester({ now: () => 0, load: () => ({ harvest: { status: 'stopped', stopReason: 'budget' } }) });
  assert.deepEqual(budgetStop.harvester.state('ann'), { state: 'idle' });
  const broken = gatedHarvester({ now: () => 0, load: () => { throw new Error('corrupt'); } });
  assert.deepEqual(broken.harvester.state('ann'), { state: 'idle' });
});

test('unknown kinds are refused', () => {
  const { harvester } = gatedHarvester({ now: () => 0 });
  for (const kind of ['constructor', 'toString', 'other'])
    assert.throws(() => harvester.request({ kind }), (error) => error.status === 400);
});

test('local fault keeps the real request count and async targets work', async (t) => {
  const original = console.error;
  console.error = () => {};
  t.after(() => { console.error = original; });
  let puts = 0;
  const cache = { isFresh: () => false, get: () => null, put() { if (++puts === 2) throw new Error('disk full'); } };
  const patches = [];
  const done = [];
  const harvester = new Harvester({
    cache, client: fakeClient(), store: { patch(value) { patches.push(value); } },
    targetsFor: async () => [{ user: 'ann', anchors: [anchor(1)], mature: false }],
    onDone: async (job, result) => { done.push(result); },
  });
  harvester.request({ kind: 'admin' });
  await settle();
  assert.deepEqual(patches.at(-1).harvest.status, 'stopped');
  assert.equal(patches.at(-1).harvest.stopReason, 'error');
  assert.equal(patches.at(-1).harvest.requests, 2);
  assert.equal(done.length, 1);
});