ca0d0d52f5
withFiles was declared inside the try block but read by the wait after it, so every check threw ReferenceError once it had imported a single model. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
286 lines
14 KiB
JavaScript
286 lines
14 KiB
JavaScript
// Watches a MakerWorld collection and imports models added to it. A check reads
|
||
// the collection page (with your token, so private collections work), imports the
|
||
// models it hasn't seen before, and remembers each one so it isn't imported twice.
|
||
// The site is read gently: a few models per check, spaced out.
|
||
const db = require('./db');
|
||
const store = require('./store');
|
||
const downloadBlock = require('./downloadBlock');
|
||
const makerworld = require('./importers/makerworld');
|
||
|
||
const IMPORT_LIMIT = 15; // models per check, so a big collection trickles in
|
||
const BETWEEN_IMPORTS_MS = 3000; // FlareSolverr runs one browser request at a time
|
||
// Downloads need much more room: MakerWorld refuses bursts a few seconds apart,
|
||
// but 67 in a row about 20 seconds apart went through without one refusal.
|
||
const BETWEEN_DOWNLOADS_MS = 20000;
|
||
const MORE_SOON_MINUTES = 10; // when a check hit the limit, come back sooner
|
||
const TICK_MS = 5 * 60 * 1000;
|
||
const INTERVALS = [1, 3, 6, 12, 24, 48];
|
||
|
||
// { fetchCollection(url), importFromUrl(url, opts), startAiRun(modelIds) }
|
||
let deps = null;
|
||
let timer = null;
|
||
let busy = false;
|
||
const checking = new Set(); // watch ids being checked right now, so checks can't overlap
|
||
|
||
const httpError = (message, status, reason) => Object.assign(new Error(message), { status, reason });
|
||
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
|
||
|
||
function init(dependencies) {
|
||
deps = dependencies;
|
||
}
|
||
|
||
// Checks anything due, every few minutes.
|
||
function start() {
|
||
// A check that was running when the server stopped isn't running now.
|
||
db.query(`UPDATE watched_collections SET last_status = NULL WHERE last_status = 'checking'`)
|
||
.catch((err) => console.warn('Could not clear stale watch checks:', err.message));
|
||
// Imports recorded before 'partial' existed: a model with an error kept none of its files.
|
||
db.query(`UPDATE watched_collection_items SET status = 'partial' WHERE status = 'imported' AND error IS NOT NULL`)
|
||
.catch((err) => console.warn('Could not mark file-less imports:', err.message));
|
||
clearInterval(timer);
|
||
timer = setInterval(() => { runDue().catch((err) => console.error('Collection watch failed:', err.message)); }, TICK_MS);
|
||
if (timer.unref) timer.unref();
|
||
runDue().catch((err) => console.error('Collection watch failed:', err.message));
|
||
}
|
||
|
||
async function list() {
|
||
const { rows } = await db.query(`
|
||
SELECT w.*, p.name AS project_name,
|
||
(SELECT COUNT(*)::int FROM watched_collection_items i WHERE i.watch_id = w.id AND i.status IN ('imported', 'partial')) AS imported_count,
|
||
(SELECT COUNT(*)::int FROM watched_collection_items i WHERE i.watch_id = w.id AND i.status = 'partial') AS partial_count,
|
||
(SELECT COUNT(*)::int FROM watched_collection_items i WHERE i.watch_id = w.id AND i.status = 'error') AS error_count,
|
||
COALESCE(r.recent, '[]'::json) AS recent,
|
||
COALESCE(q.pending, '[]'::json) AS pending
|
||
FROM watched_collections w
|
||
LEFT JOIN projects p ON p.id = w.project_id
|
||
LEFT JOIN LATERAL (
|
||
SELECT json_agg(json_build_object('design_id', i.design_id, 'model_id', i.model_id, 'title', i.title,
|
||
'status', i.status, 'error', i.error, 'seen_at', i.seen_at)
|
||
ORDER BY i.seen_at DESC) AS recent
|
||
FROM (SELECT * FROM watched_collection_items WHERE watch_id = w.id ORDER BY seen_at DESC LIMIT 5) i
|
||
) r ON true
|
||
-- Every model still waiting for its files, not just the five most recent.
|
||
LEFT JOIN LATERAL (
|
||
SELECT json_agg(json_build_object('model_id', i.model_id, 'title', i.title) ORDER BY i.seen_at DESC) AS pending
|
||
FROM (SELECT * FROM watched_collection_items
|
||
WHERE watch_id = w.id AND status = 'partial' AND model_id IS NOT NULL
|
||
ORDER BY seen_at DESC LIMIT 200) i
|
||
) q ON true
|
||
ORDER BY w.id`);
|
||
return rows;
|
||
}
|
||
|
||
async function get(id) {
|
||
const { rows: [watch] } = await db.query(`
|
||
SELECT w.*, p.name AS project_name,
|
||
(SELECT COUNT(*)::int FROM watched_collection_items i WHERE i.watch_id = w.id AND i.status IN ('imported', 'partial')) AS imported_count,
|
||
(SELECT COUNT(*)::int FROM watched_collection_items i WHERE i.watch_id = w.id AND i.status = 'partial') AS partial_count,
|
||
(SELECT COUNT(*)::int FROM watched_collection_items i WHERE i.watch_id = w.id AND i.status = 'error') AS error_count
|
||
FROM watched_collections w LEFT JOIN projects p ON p.id = w.project_id WHERE w.id = $1`, [id]);
|
||
if (!watch) return null;
|
||
const { rows: recent } = await db.query(`
|
||
SELECT i.design_id, i.model_id, i.title, i.status, i.error, i.seen_at, m.name AS model_name
|
||
FROM watched_collection_items i LEFT JOIN models m ON m.id = i.model_id
|
||
WHERE i.watch_id = $1 ORDER BY i.seen_at DESC LIMIT 12`, [id]);
|
||
return { ...watch, recent };
|
||
}
|
||
|
||
// Adds a watch. Reads the collection first, so a bad link or a refused token is
|
||
// reported now rather than silently later.
|
||
async function add({ url, aiTag = false, intervalHours = 6, fetchFiles = false }) {
|
||
const sourceId = makerworld.collectionIdFromUrl(url);
|
||
if (!sourceId) throw httpError('That isn’t a MakerWorld collection link (makerworld.com/…/collections/…).', 400);
|
||
const { rowCount } = await db.query('SELECT 1 FROM watched_collections WHERE site = $1 AND source_id = $2', ['makerworld', sourceId]);
|
||
if (rowCount) throw httpError('You’re already watching that collection.', 409);
|
||
|
||
const collection = await deps.fetchCollection(url);
|
||
const project = await store.findOrCreateProject(collection.title || `MakerWorld collection ${sourceId}`);
|
||
const hours = INTERVALS.includes(Number(intervalHours)) ? Number(intervalHours) : 6;
|
||
const { rows: [watch] } = await db.query(
|
||
`INSERT INTO watched_collections (site, source_id, url, title, project_id, ai_tag, interval_hours, total, listed, last_status, fetch_files)
|
||
VALUES ('makerworld', $1, $2, $3, $4, $5, $6, $7, $8, 'checking', $9) RETURNING id`,
|
||
[sourceId, url, collection.title, project.id, Boolean(aiTag), hours, collection.total, collection.designs.length,
|
||
Boolean(fetchFiles)]);
|
||
// The collection was just read, so import straight from it rather than
|
||
// fetching the whole thing again.
|
||
const added = await get(watch.id);
|
||
checking.add(watch.id);
|
||
importFound(added, collection)
|
||
.catch((err) => console.warn(`Watch ${watch.id} first import failed:`, err.message))
|
||
.finally(() => checking.delete(watch.id));
|
||
return added;
|
||
}
|
||
|
||
async function update(id, { enabled, aiTag, intervalHours, fetchFiles }) {
|
||
const hours = INTERVALS.includes(Number(intervalHours)) ? Number(intervalHours) : null;
|
||
const { rowCount } = await db.query(
|
||
`UPDATE watched_collections SET
|
||
enabled = COALESCE($2, enabled),
|
||
ai_tag = COALESCE($3, ai_tag),
|
||
interval_hours = COALESCE($4, interval_hours),
|
||
fetch_files = COALESCE($5, fetch_files),
|
||
next_check_at = CASE WHEN $2 IS TRUE THEN now() ELSE next_check_at END
|
||
WHERE id = $1`,
|
||
[id, enabled === undefined ? null : Boolean(enabled), aiTag === undefined ? null : Boolean(aiTag), hours,
|
||
fetchFiles === undefined ? null : Boolean(fetchFiles)]);
|
||
if (!rowCount) return null;
|
||
if (enabled) runDue().catch(() => {});
|
||
return get(id);
|
||
}
|
||
|
||
// Stops watching. Models already imported stay in the library.
|
||
async function remove(id) {
|
||
const { rowCount } = await db.query('DELETE FROM watched_collections WHERE id = $1', [id]);
|
||
return rowCount > 0;
|
||
}
|
||
|
||
// Checks one collection now: reads it (awaited, so problems come back to the
|
||
// caller), then imports in the background.
|
||
async function checkNow(id) {
|
||
const watch = await get(id);
|
||
if (!watch) throw httpError('Not watching that collection.', 404);
|
||
if (checking.has(id)) return watch; // already going; don't import anything twice
|
||
let collection;
|
||
try {
|
||
collection = await deps.fetchCollection(watch.url);
|
||
} catch (err) {
|
||
await recordFailure(watch, err);
|
||
throw err;
|
||
}
|
||
await db.query(
|
||
`UPDATE watched_collections SET title = $2, total = $3, listed = $4, last_status = 'checking',
|
||
last_error = NULL, last_error_reason = NULL WHERE id = $1`,
|
||
[id, collection.title || watch.title, collection.total, collection.designs.length]);
|
||
checking.add(id);
|
||
importFound(watch, collection)
|
||
.catch((err) => console.error(`Watch ${id} import failed:`, err.message))
|
||
.finally(() => checking.delete(id));
|
||
return get(id);
|
||
}
|
||
|
||
async function recordFailure(watch, err) {
|
||
await db.query(
|
||
`UPDATE watched_collections SET last_checked_at = now(), last_status = 'error', last_error = $2,
|
||
last_error_reason = $3, next_check_at = now() + make_interval(hours => GREATEST(interval_hours, 1))
|
||
WHERE id = $1`,
|
||
[watch.id, String(err.message).slice(0, 300), err.reason || null]);
|
||
}
|
||
|
||
// Imports the models this watch hasn't seen yet, a few at a time.
|
||
async function importFound(watch, collection) {
|
||
const { rows: seenRows } = await db.query(
|
||
`SELECT design_id FROM watched_collection_items WHERE watch_id = $1 AND status <> 'error'`, [watch.id]);
|
||
const seen = new Set(seenRows.map((r) => r.design_id));
|
||
const pending = collection.designs.filter((d) => !seen.has(d.id));
|
||
const todo = pending.slice(0, IMPORT_LIMIT);
|
||
|
||
let added = 0;
|
||
let problem = null;
|
||
let fileMisses = 0; // downloads refused in a row
|
||
let stoppedEarly = false; // left models unimported, so come back soon
|
||
const fresh = [];
|
||
for (const design of todo) {
|
||
// Paused while this batch was running: stop, rather than import the rest.
|
||
const { rows: [live] } = await db.query('SELECT enabled FROM watched_collections WHERE id = $1', [watch.id]);
|
||
if (!live || !live.enabled) break;
|
||
const url = makerworld.designUrl(design);
|
||
// While MakerWorld is waiting for you to prove you're not a robot, the
|
||
// check still brings in the models — their files wait for you. Decided out
|
||
// here because the wait after each import depends on it.
|
||
const withFiles = Boolean(watch.fetch_files) && !downloadBlock.isBlocked('makerworld');
|
||
try {
|
||
const result = await deps.importFromUrl(url, {
|
||
projects: watch.project_id ? [watch.project_id] : [],
|
||
withFiles, // off by default: MakerWorld refuses quick bursts
|
||
});
|
||
if (result.existing) {
|
||
if (watch.project_id) await store.addModelToProject(result.model.id, watch.project_id);
|
||
await recordItem(watch.id, design, result.model.id, 'existing', null);
|
||
} else {
|
||
added += 1;
|
||
fresh.push(result.model.id);
|
||
// Without files (or when they couldn't be fetched) the model is only
|
||
// 'partial' — "Get files" fills it in later.
|
||
const gotFiles = withFiles && !result.filesError;
|
||
await recordItem(watch.id, design, result.model.id, gotFiles ? 'imported' : 'partial',
|
||
result.filesError || null);
|
||
}
|
||
// Refused downloads mean we're asking too fast. Stop the batch and come
|
||
// back shortly, rather than collecting refusals for the whole of it.
|
||
if (withFiles && result.filesError) {
|
||
fileMisses += 1;
|
||
if (fileMisses >= 2 && result.filesErrorReason !== 'makerworld_token') { stoppedEarly = true; break; }
|
||
} else if (!result.filesError) {
|
||
fileMisses = 0;
|
||
}
|
||
// A refused token, or a check only you can solve, means every later
|
||
// download fails too: stop and say so.
|
||
if (['makerworld_token', 'makerworld_challenge'].includes(result.filesErrorReason)) {
|
||
problem = httpError(result.filesError, 502, result.filesErrorReason);
|
||
break;
|
||
}
|
||
} catch (err) {
|
||
await recordItem(watch.id, design, null, 'error', err.message);
|
||
if (['makerworld_token', 'makerworld_challenge'].includes(err.reason)) { problem = err; break; }
|
||
}
|
||
await sleep(withFiles ? BETWEEN_DOWNLOADS_MS : BETWEEN_IMPORTS_MS);
|
||
}
|
||
|
||
const more = pending.length > todo.length || stoppedEarly;
|
||
if (problem) {
|
||
await recordFailure({ ...watch, id: watch.id }, problem);
|
||
} else {
|
||
await db.query(
|
||
`UPDATE watched_collections SET last_checked_at = now(), last_status = 'ok', last_error = NULL,
|
||
last_error_reason = NULL, last_added = $2,
|
||
next_check_at = now() + CASE WHEN $3::boolean THEN make_interval(mins => $4) ELSE make_interval(hours => interval_hours) END
|
||
WHERE id = $1`,
|
||
[watch.id, added, more, MORE_SOON_MINUTES]);
|
||
}
|
||
|
||
if (watch.ai_tag && fresh.length && deps.startAiRun) {
|
||
await deps.startAiRun(fresh).catch((err) => console.warn(`Watch ${watch.id}: AI tagging didn't start:`, err.message));
|
||
}
|
||
}
|
||
|
||
async function recordItem(watchId, design, modelId, status, error) {
|
||
await db.query(
|
||
`INSERT INTO watched_collection_items (watch_id, design_id, model_id, title, status, error, seen_at)
|
||
VALUES ($1, $2, $3, $4, $5, $6, now())
|
||
ON CONFLICT (watch_id, design_id) DO UPDATE SET model_id = EXCLUDED.model_id, title = EXCLUDED.title,
|
||
status = EXCLUDED.status, error = EXCLUDED.error, seen_at = now()`,
|
||
[watchId, design.id, modelId, String(design.title || '').slice(0, 200), status, error ? String(error).slice(0, 300) : null]);
|
||
}
|
||
|
||
// Checks every watch that's due, one at a time.
|
||
async function runDue() {
|
||
if (busy || !deps) return;
|
||
busy = true;
|
||
try {
|
||
for (;;) {
|
||
const { rows: [due] } = await db.query(
|
||
`SELECT id FROM watched_collections WHERE enabled AND next_check_at <= now()
|
||
AND id <> ALL($1::int[]) ORDER BY next_check_at LIMIT 1`, [[...checking]]);
|
||
if (!due) return;
|
||
// checkNow leaves the import running in the background; wait for it here so
|
||
// watches are checked one after another.
|
||
const watch = await get(due.id);
|
||
checking.add(watch.id);
|
||
try {
|
||
const collection = await deps.fetchCollection(watch.url);
|
||
await db.query(
|
||
`UPDATE watched_collections SET title = $2, total = $3, listed = $4, last_status = 'checking' WHERE id = $1`,
|
||
[watch.id, collection.title || watch.title, collection.total, collection.designs.length]);
|
||
await importFound(watch, collection);
|
||
} catch (err) {
|
||
await recordFailure(watch, err);
|
||
} finally {
|
||
checking.delete(watch.id);
|
||
}
|
||
}
|
||
} finally {
|
||
busy = false;
|
||
}
|
||
}
|
||
|
||
module.exports = { INTERVALS, init, start, list, get, add, update, remove, checkNow, runDue };
|