cd95890296
Models imported before 'partial' existed are stored as 'imported' with an error, so they'd never show the button. Mark them on startup, and key the button on the stored error rather than the status. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
243 lines
12 KiB
JavaScript
243 lines
12 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 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
|
||
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
|
||
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
|
||
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 }) {
|
||
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)
|
||
VALUES ('makerworld', $1, $2, $3, $4, $5, $6, $7, $8, 'checking') RETURNING id`,
|
||
[sourceId, url, collection.title, project.id, Boolean(aiTag), hours, collection.total, collection.designs.length]);
|
||
checkNow(watch.id).catch((err) => console.warn(`Watch ${watch.id} first check failed:`, err.message));
|
||
return get(watch.id);
|
||
}
|
||
|
||
async function update(id, { enabled, aiTag, intervalHours }) {
|
||
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),
|
||
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]);
|
||
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;
|
||
const fresh = [];
|
||
for (const design of todo) {
|
||
const url = makerworld.designUrl(design);
|
||
try {
|
||
const result = await deps.importFromUrl(url, { projects: watch.project_id ? [watch.project_id] : [] });
|
||
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);
|
||
// Imported, but MakerWorld wouldn't hand over the files: say so rather
|
||
// than letting an empty model look like a success.
|
||
await recordItem(watch.id, design, result.model.id,
|
||
result.filesError ? 'partial' : 'imported', result.filesError || null);
|
||
}
|
||
// A refused token means every later download fails too: stop and say so.
|
||
if (result.filesErrorReason === 'makerworld_token') {
|
||
problem = httpError(result.filesError, 502, 'makerworld_token');
|
||
break;
|
||
}
|
||
} catch (err) {
|
||
await recordItem(watch.id, design, null, 'error', err.message);
|
||
if (err.reason === 'makerworld_token') { problem = err; break; }
|
||
}
|
||
await sleep(BETWEEN_IMPORTS_MS);
|
||
}
|
||
|
||
const more = pending.length > todo.length;
|
||
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 };
|