spb/tools-spboucher-ai

1/** Job handlers (run by the worker). Each is idempotent and writes its own evidence rows. */2import { sql } from "@/db";3import { safeFetch, extractHtmlMeta, normalizeUrl, sameSite } from "@/lib/url";4import { getEmbeddingProvider } from "@/lib/ai/provider";5import { enqueue, type Job } from "./queue";6import { createHash } from "node:crypto";78const URL_FIELDS = ["homepage_url", "docs_url", "repo_url", "pricing_url", "license_url", "privacy_url", "security_url"] as const;910/** Check every URL of one tool; record link_checks rows; update tools.link_state/checked_at; refresh sources.http_status. */11export async function linkCheck(toolId: string) {12  const [t] = await sql<any[]>`SELECT id, slug, ${sql.unsafe(URL_FIELDS.join(", "))} FROM tools WHERE id = ${toolId}`;13  if (!t) return { skipped: "missing" };14  let worst: "ok" | "redirected" | "broken" = "ok"; const results: any[] = [];15  for (const f of URL_FIELDS) {16    const url = t[f]; if (!url) continue;17    const r = await safeFetch(url, { method: "GET", maxBytes: 262_144, timeoutMs: 15_000 });18    // Some sites reject HEAD/bots with 403/405/429 while being alive: treat as "unverifiable" (ok=false but not broken) when status is 401/403/405/429.19    // Bot mitigation shows up as 401/403/405/429, as self-redirect loops, or as connections that stall until the timeout.20    // These are "unverifiable", not evidence that the page is gone; 404/410/5xx/DNS failures are.21    const soft = [401, 403, 405, 429, 999].includes(r.status) || r.error === "timeout" || (r.error ?? "").startsWith("self-redirect") || r.error === "final URL is not https";22    const ok = r.ok || soft;23    const offsite = r.redirected && !sameSite(url, r.finalUrl);24    await sql`INSERT INTO link_checks (tool_id, field, url, ok, status_code, final_url, redirected, error, duration_ms) VALUES (${toolId}, ${f}, ${url}, ${ok}, ${r.status || null}, ${r.finalUrl}, ${r.redirected}, ${soft ? `soft:${r.error ?? r.status}` : (r.error ?? null)}, ${r.durationMs})`;25    if (!ok) worst = "broken"; else if (offsite && worst === "ok") worst = "redirected";26    results.push({ field: f, status: r.status, ok, redirected: r.redirected, finalUrl: r.finalUrl, error: r.error });27    await sql`UPDATE sources SET http_status = ${r.status || null}, final_url = ${r.finalUrl}, fetched_at = now(), fetch_error = ${r.error ?? null}, updated_at = now() WHERE url = ${normalizeUrl(url)}`.catch(() => {});28    await new Promise((res) => setTimeout(res, 250)); // politeness29  }30  await sql`UPDATE tools SET link_state = ${worst}, checked_at = now(), updated_at = now() WHERE id = ${toolId}`;31  return { linkState: worst, results };32}3334/** Re-fetch the official site of a tool, store title/canonical/hash on the source row, and flag a diff for editorial review when the hash changed. */35export async function sourceRescan(toolId: string) {36  const [t] = await sql<any[]>`SELECT id, slug, homepage_url FROM tools WHERE id = ${toolId}`; if (!t) return { skipped: "missing" };37  const r = await safeFetch(t.homepage_url, { maxBytes: 1_048_576 });38  const meta = r.body ? extractHtmlMeta(r.body) : { title: null, canonical: null, description: null };39  const hash = r.body ? createHash("sha256").update(r.body.replace(/\s+/g, " ")).digest("hex") : null;40  const url = normalizeUrl(t.homepage_url);41  const [prev] = await sql<any[]>`SELECT content_hash, canonical_url FROM sources WHERE url = ${url}`;42  await sql`INSERT INTO sources (url, source_type, title, canonical_url, http_status, final_url, content_hash, fetched_at, fetch_error) VALUES (${url}, 'official_site', ${meta.title}, ${meta.canonical}, ${r.status || null}, ${r.finalUrl}, ${hash}, now(), ${r.error ?? null})43    ON CONFLICT (url) DO UPDATE SET title = coalesce(EXCLUDED.title, sources.title), canonical_url = EXCLUDED.canonical_url, http_status = EXCLUDED.http_status, final_url = EXCLUDED.final_url, content_hash = EXCLUDED.content_hash, fetched_at = now(), fetch_error = EXCLUDED.fetch_error, updated_at = now()`;44  const changed = !!prev?.content_hash && !!hash && prev.content_hash !== hash;45  if (changed) await sql`INSERT INTO settings (key, value) VALUES (${"review:" + t.slug}, ${sql.json({ kind: "source_changed", url, title: meta.title, canonical: meta.canonical, at: new Date().toISOString() })}) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`;46  if (meta.canonical && !sameSite(meta.canonical, t.homepage_url)) await sql`INSERT INTO settings (key, value) VALUES (${"review:" + t.slug + ":canonical"}, ${sql.json({ kind: "canonical_moved", from: url, to: meta.canonical, at: new Date().toISOString() })}) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`;47  return { status: r.status, title: meta.title, canonical: meta.canonical, changed };48}4950export async function embedTool(toolId: string) {51  const p = getEmbeddingProvider(); if (!p) return { skipped: "no provider" };52  const [t] = await sql<any[]>`SELECT id, embedding_text, content_hash, embedding_model FROM tools WHERE id = ${toolId}`; if (!t?.embedding_text) return { skipped: "no text" };53  const tag = `${p.model}:${t.content_hash}`; if (t.embedding_model === tag) return { skipped: "up to date" };54  const [v] = await p.embed([t.embedding_text.slice(0, 8000)]);55  await sql`UPDATE tools SET embedding = ${JSON.stringify(v)}::vector, embedding_model = ${tag}, updated_at = updated_at WHERE id = ${toolId}`;56  return { model: p.model };57}5859/** Schedulers: fan out per-tool jobs with dedupe keys. */60export async function linkSweep() {61  const rows = await sql<{ id: string }[]>`SELECT id FROM tools WHERE visibility='public' AND editorial_status IN ('published','reviewed') AND (checked_at IS NULL OR checked_at < now() - interval '6 days') ORDER BY checked_at NULLS FIRST LIMIT 400`;62  for (const [i, r] of rows.entries()) await enqueue("link_check", { toolId: r.id }, { dedupeKey: `link_check:${r.id}`, runAt: new Date(Date.now() + i * 3000), priority: 7 });63  return { enqueued: rows.length };64}65export async function embedSweep() {66  if (!getEmbeddingProvider()) return { skipped: "no provider" };67  const rows = await sql<{ id: string }[]>`SELECT id FROM tools WHERE embedding_text IS NOT NULL AND (embedding IS NULL OR embedding_model IS DISTINCT FROM ${""} ) LIMIT 500`;68  let n = 0; for (const r of rows) { const [t] = await sql<any[]>`SELECT content_hash, embedding_model FROM tools WHERE id=${r.id}`; const p = getEmbeddingProvider()!; if (t.embedding_model !== `${p.model}:${t.content_hash}`) { await enqueue("embed_tool", { toolId: r.id }, { dedupeKey: `embed:${r.id}`, priority: 8 }); n++; } }69  return { enqueued: n };70}71/** Freshness policy by field: pricing 90 d, license 365 d, review 365 d, sources 180 d → flags in settings `stale:<slug>` for the admin review queue. */72export async function freshnessSweep() {73  const rows = await sql<any[]>`74    SELECT t.slug, t.pricing_checked_at, t.reviewed_at, t.pricing_url, (SELECT min(checked_at) FROM tool_sources ts WHERE ts.tool_id = t.id) AS oldest_source75    FROM tools t WHERE t.visibility='public' AND t.editorial_status='published'`;76  let flagged = 0;77  for (const r of rows) {78    const reasons: string[] = [];79    if (r.pricing_url && (!r.pricing_checked_at || Date.now() - new Date(r.pricing_checked_at).getTime() > 90 * 864e5)) reasons.push("pricing older than 90 days");80    if (r.reviewed_at && Date.now() - new Date(r.reviewed_at).getTime() > 365 * 864e5) reasons.push("editorial review older than 365 days");81    if (r.oldest_source && Date.now() - new Date(r.oldest_source).getTime() > 180 * 864e5) reasons.push("a source was last checked over 180 days ago");82    if (reasons.length) { flagged++; await sql`INSERT INTO settings (key, value) VALUES (${"stale:" + r.slug}, ${sql.json({ kind: "stale", reasons, at: new Date().toISOString() })}) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`; }83    else await sql`DELETE FROM settings WHERE key = ${"stale:" + r.slug}`;84  }85  return { flagged, total: rows.length };86}87export async function qualityRecompute() {88  // quality_score depends on link_state: recompute the link component only (content component set at import/edit)89  const r = await sql`UPDATE tools SET quality_score = LEAST(100, GREATEST(0, quality_score)) WHERE true`; return { rows: r.count };90}9192export async function runJob(job: Job): Promise<unknown> {93  const id = String(job.payload.toolId ?? "");94  switch (job.kind) {95    case "link_check": return linkCheck(id);96    case "link_sweep": return linkSweep();97    case "source_rescan": return sourceRescan(id);98    case "embed_tool": return embedTool(id);99    case "embed_sweep": return embedSweep();100    case "freshness_sweep": return freshnessSweep();101    case "quality_recompute": return qualityRecompute();102    case "backup": { const { runBackup } = await import("./backup"); return runBackup(); }103    case "import_batch": { const { importAll } = await import("@/lib/content/import"); return importAll({ root: String(job.payload.root), publish: !!job.payload.publish, actor: String(job.payload.actor ?? "worker") }); }104    default: throw new Error(`unknown job kind ${job.kind}`);105  }106}107