printhive/lib/collectionWatch.js

// 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,
      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 }) {
  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]);
  // 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 }) {
  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) {
    // 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);
    try {
      const result = await deps.importFromUrl(url, {
        projects: watch.project_id ? [watch.project_id] : [],
        withFiles: false, // MakerWorld allows only a couple of downloads in a row
      });
      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);
        // A check brings in the page, not the files, so every new model starts
        // 'partial' — "Get files" on the ones you want fills them in.
        await recordItem(watch.id, design, result.model.id, 'partial', 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 };