TypeScript 97.5%
SQL 1.4%
Python 0.8%
1import {2 FetchaError,3 PLAN_LIMITS,4 assertUrlAllowed,5 crawlCreateSchema,6 extractDomain,7 extractPageMetadata,8 looksLikePage,9 mapCreateSchema,10 newId,11 normalizeCrawlUrl,12 parseRobots,13 parseSitemap,14 registrableHost,15 robotsAllows,16 urlPatternMatcher,17 type CrawlCreateInput,18 type FetchResponseBody,19 type MapCreateInput,20 type RobotsRules,21} from "@fetcha/core";22import { and, crawlJobs, crawlPages, db, desc, eq, inArray, lt, sql, type CrawlJob, type CrawlPage } from "@fetcha/db";23import { hostname } from "node:os";24import type { ApiPrincipal } from "../auth";25import { handleFetch, type FetchHandlerOutput } from "../routes/fetch";2627/**28 * Crawl service: durable jobs (crawl_jobs / crawl_pages) executed in-process by a small worker29 * loop. Every page goes through `handleFetch`, so quotas, routing intelligence, block detection and30 * browser escalation apply exactly as for a single fetch.31 */3233const WORKER_ID = `${hostname()}:${process.pid}`;34const PAGE_CONTENT_CAP = 600_000; // chars stored per page35const HEARTBEAT_MS = 10_000;36const STALE_MS = 3 * 60_000;3738export function serializeJob(j: CrawlJob) {39 return {40 id: j.id,41 status: j.status,42 label: j.label,43 seed_url: j.seedUrl,44 domain: j.domain,45 source: j.source,46 options: j.options,47 stats: { discovered: j.pagesDiscovered, fetched: j.pagesFetched, ok: j.pagesOk, blocked: j.pagesBlocked, failed: j.pagesFailed, bytes: j.bytes },48 error: j.errorCode ? { code: j.errorCode, message: j.errorMessage } : null,49 webhook_status: j.webhookStatus,50 created_at: j.createdAt.toISOString(),51 started_at: j.startedAt?.toISOString() ?? null,52 completed_at: j.completedAt?.toISOString() ?? null,53 };54}5556export function serializePage(p: CrawlPage, includeContent = true) {57 return {58 id: p.id,59 url: p.url,60 final_url: p.finalUrl,61 depth: p.depth,62 parent_url: p.parentUrl,63 status: p.status,64 http_status: p.httpStatus,65 error_code: p.errorCode,66 request_id: p.requestId,67 title: p.title,68 description: p.description,69 content_type: p.contentType,70 content: includeContent ? p.content : undefined,71 links_count: p.linksCount,72 bytes: p.bytes,73 duration_ms: p.durationMs,74 mode: p.mode,75 fetched_at: p.fetchedAt?.toISOString() ?? null,76 };77}7879// ---------------------------------------------------------------------------80// Public operations81// ---------------------------------------------------------------------------82export async function createCrawl(principal: ApiPrincipal, body: unknown, source: "api" | "playground" | "sdk") {83 const parsed = crawlCreateSchema.safeParse(body ?? {});84 if (!parsed.success) {85 throw new FetchaError("INVALID_REQUEST", "Invalid crawl options.", { details: { issues: parsed.error.issues.map((i) => ({ path: i.path.join("."), message: i.message })) } });86 }87 const opts = parsed.data;88 const limits = PLAN_LIMITS[principal.plan];89 if (opts.max_pages > limits.crawl_max_pages) opts.max_pages = limits.crawl_max_pages;90 if (opts.timeout > limits.max_timeout_ms) opts.timeout = limits.max_timeout_ms;91 if (opts.network !== "auto" && !limits.networks.includes(opts.network)) throw new FetchaError("NETWORK_UNAVAILABLE");92 const allowed = await assertUrlAllowed(opts.url);93 const seed = normalizeCrawlUrl(allowed.url.toString());94 if (!seed) throw new FetchaError("INVALID_REQUEST", "The seed URL is invalid.");95 if (opts.webhook_url) await assertUrlAllowed(opts.webhook_url);9697 const [active] = await db98 .select({ n: sql<number>`count(*)::int` })99 .from(crawlJobs)100 .where(and(eq(crawlJobs.organizationId, principal.organizationId), inArray(crawlJobs.status, ["queued", "running"])));101 if ((active?.n ?? 0) >= limits.crawl_concurrent_jobs) {102 throw new FetchaError("CRAWL_LIMIT_REACHED", `At most ${limits.crawl_concurrent_jobs} crawl jobs can be queued or running at once.`, { details: { limit: limits.crawl_concurrent_jobs } });103 }104105 const id = newId("crawl");106 await db.insert(crawlJobs).values({107 id,108 organizationId: principal.organizationId,109 projectId: principal.projectId,110 apiKeyId: principal.keyId,111 source,112 label: opts.label ?? null,113 seedUrl: seed,114 domain: extractDomain(seed),115 options: { ...opts, url: seed } as Record<string, unknown>,116 status: "queued",117 pagesDiscovered: 1,118 });119 await db.insert(crawlPages).values({ id: newId("cpg"), jobId: id, url: seed, depth: 0, status: "pending" }).onConflictDoNothing();120 const [job] = await db.select().from(crawlJobs).where(eq(crawlJobs.id, id)).limit(1);121 kick();122 return serializeJob(job!);123}124125export async function getCrawl(principal: ApiPrincipal, id: string) {126 const [job] = await db.select().from(crawlJobs).where(and(eq(crawlJobs.id, id), eq(crawlJobs.projectId, principal.projectId))).limit(1);127 if (!job) throw new FetchaError("CRAWL_NOT_FOUND");128 return serializeJob(job);129}130131export async function listCrawls(principal: ApiPrincipal, limit = 50) {132 const rows = await db.select().from(crawlJobs).where(eq(crawlJobs.projectId, principal.projectId)).orderBy(desc(crawlJobs.createdAt)).limit(Math.min(Math.max(1, limit), 200));133 return { data: rows.map(serializeJob) };134}135136export async function listCrawlPages(principal: ApiPrincipal, id: string, q: { cursor?: string; limit?: number; status?: string; include_content?: boolean }) {137 const [job] = await db.select({ id: crawlJobs.id }).from(crawlJobs).where(and(eq(crawlJobs.id, id), eq(crawlJobs.projectId, principal.projectId))).limit(1);138 if (!job) throw new FetchaError("CRAWL_NOT_FOUND");139 const limit = Math.min(Math.max(1, q.limit ?? 100), 500);140 const conds = [eq(crawlPages.jobId, id)];141 if (q.status) conds.push(eq(crawlPages.status, q.status));142 else conds.push(sql`${crawlPages.status} <> 'pending'`);143 if (q.cursor) {144 const [ts, cid] = decodeCursor(q.cursor);145 conds.push(sql`(${crawlPages.createdAt}, ${crawlPages.id}) > (${new Date(ts)}, ${cid})`);146 }147 const rows = await db148 .select()149 .from(crawlPages)150 .where(and(...conds))151 .orderBy(crawlPages.createdAt, crawlPages.id)152 .limit(limit + 1);153 const page = rows.slice(0, limit);154 const last = page.at(-1);155 return { data: page.map((p) => serializePage(p, q.include_content !== false)), next_cursor: rows.length > limit && last ? encodeCursor(last.createdAt, last.id) : null };156}157158export async function cancelCrawl(principal: ApiPrincipal, id: string) {159 const res = await db160 .update(crawlJobs)161 .set({ status: "cancelled", completedAt: new Date() })162 .where(and(eq(crawlJobs.id, id), eq(crawlJobs.projectId, principal.projectId), inArray(crawlJobs.status, ["queued", "running"])))163 .returning({ id: crawlJobs.id });164 if (!res.length) {165 const [job] = await db.select({ id: crawlJobs.id, status: crawlJobs.status }).from(crawlJobs).where(and(eq(crawlJobs.id, id), eq(crawlJobs.projectId, principal.projectId))).limit(1);166 if (!job) throw new FetchaError("CRAWL_NOT_FOUND");167 return { id, status: job.status };168 }169 cancelled.add(id);170 return { id, status: "cancelled" };171}172173function encodeCursor(d: Date, id: string): string {174 return Buffer.from(`${d.toISOString()}|${id}`).toString("base64url");175}176function decodeCursor(c: string): [string, string] {177 try {178 const [ts, id] = Buffer.from(c, "base64url").toString("utf8").split("|");179 if (!ts || !id || Number.isNaN(Date.parse(ts))) throw new Error("bad");180 return [ts, id];181 } catch {182 throw new FetchaError("INVALID_REQUEST", "Invalid cursor.");183 }184}185186// ---------------------------------------------------------------------------187// Map (sync)188// ---------------------------------------------------------------------------189export async function mapSite(principal: ApiPrincipal, body: unknown, source: "api" | "playground" | "sdk", clientIp: string | null) {190 const parsed = mapCreateSchema.safeParse(body ?? {});191 if (!parsed.success) {192 throw new FetchaError("INVALID_REQUEST", "Invalid map options.", { details: { issues: parsed.error.issues.map((i) => ({ path: i.path.join("."), message: i.message })) } });193 }194 const opts = parsed.data;195 const allowed = await assertUrlAllowed(opts.url);196 const seed = normalizeCrawlUrl(allowed.url.toString())!;197 const seedUrl = new URL(seed);198 const host = registrableHost(seedUrl.hostname);199 const inScope = scopeFilter(seedUrl, opts.same_domain, opts.allow_subdomains);200 const search = opts.search ? urlPatternMatcher([opts.search.includes("*") || /^\/.+\/[a-z]*$/.test(opts.search) ? opts.search : `*${opts.search}*`]) : null;201 const deadline = Date.now() + Math.min(55_000, opts.timeout * 2);202 const found = new Map<string, "sitemap" | "links">();203 const add = (u: string, src: "sitemap" | "links") => {204 const n = normalizeCrawlUrl(u);205 if (!n || found.has(n) || !inScope(n) || !looksLikePage(n)) return;206 if (search && !search(n)) return;207 found.set(n, src);208 };209 const fetchText = async (url: string, format: "raw" | "html" = "raw", extra: Record<string, unknown> = {}): Promise<FetchResponseBody | null> => {210 if (Date.now() > deadline - 2000) return null;211 try {212 const out = await handleFetch({ principal, body: { url, format, timeout: Math.min(opts.timeout, Math.max(3000, deadline - Date.now() - 1000)), country: opts.country, network: opts.network, browser_fallback: false, retries: 1, ...extra }, source, clientIp, userAgent: `fetcha-map/${source}` });213 return out.body as FetchResponseBody;214 } catch {215 return null;216 }217 };218 let sitemapCount = 0;219 let linksCount = 0;220 if (opts.use_sitemap) {221 const sitemapUrls = new Set<string>([`${seedUrl.origin}/sitemap.xml`, `${seedUrl.origin}/sitemap_index.xml`]);222 const robots = await fetchText(`${seedUrl.origin}/robots.txt`);223 if (robots?.success && robots.content) for (const s of parseRobots(robots.content).sitemaps) sitemapUrls.add(s);224 const queue = [...sitemapUrls];225 const seen = new Set<string>();226 while (queue.length && seen.size < 25 && found.size < opts.limit && Date.now() < deadline - 3000) {227 const u = queue.shift()!;228 if (seen.has(u)) continue;229 seen.add(u);230 const r = await fetchText(u);231 if (!r?.success || !r.content || !/<(urlset|sitemapindex)|<loc>/i.test(r.content.slice(0, 5000)) && !/^https?:\/\//m.test(r.content.slice(0, 500))) continue;232 const { urls, sitemaps } = parseSitemap(r.content);233 for (const s of sitemaps) if (host === safeHost(s) || opts.allow_subdomains) queue.push(s);234 for (const x of urls) {235 if (found.size >= opts.limit) break;236 const before = found.size;237 add(x, "sitemap");238 if (found.size > before) sitemapCount++;239 }240 }241 }242 if (opts.use_links && found.size < opts.limit && Date.now() < deadline - 3000) {243 const r = await fetchText(seed, "html", { links: true });244 if (r?.success && r.links) {245 for (const l of r.links) {246 if (found.size >= opts.limit) break;247 const before = found.size;248 add(l.url, "links");249 if (found.size > before) linksCount++;250 }251 } else if (r?.content) {252 const meta = extractPageMetadata(r.content, r.final_url);253 for (const l of meta.links) {254 if (found.size >= opts.limit) break;255 const before = found.size;256 add(l.url, "links");257 if (found.size > before) linksCount++;258 }259 }260 }261 const urls = [...found.keys()];262 return { url: seed, count: urls.length, urls, sources: { sitemap: sitemapCount, links: linksCount }, truncated: urls.length >= opts.limit };263}264265function safeHost(u: string): string {266 try {267 return registrableHost(new URL(u).hostname);268 } catch {269 return "";270 }271}272273function scopeFilter(seed: URL, sameDomain: boolean, allowSubdomains: boolean): (url: string) => boolean {274 const seedHost = seed.hostname.toLowerCase().replace(/^www\./, "");275 const reg = registrableHost(seed.hostname);276 return (url: string) => {277 if (!sameDomain) return true;278 let h: string;279 try {280 h = new URL(url).hostname.toLowerCase().replace(/^www\./, "");281 } catch {282 return false;283 }284 if (h === seedHost) return true;285 if (allowSubdomains) return registrableHost(h) === reg;286 return false;287 };288}289290// ---------------------------------------------------------------------------291// Worker292// ---------------------------------------------------------------------------293const cancelled = new Set<string>();294let running = false;295let stopped = false;296let wake: (() => void) | null = null;297const activeJobs = new Map<string, Promise<void>>();298const MAX_PARALLEL_JOBS = Number(process.env.FETCHA_CRAWL_PARALLEL_JOBS ?? 4);299300function kick() {301 wake?.();302}303304export function startCrawlWorker(log: { info: (m: string) => void; warn: (m: string) => void }): () => Promise<void> {305 if (running) return async () => {};306 running = true;307 stopped = false;308 const loop = async () => {309 // Re-queue jobs left "running" by a previous process (crash / redeploy).310 await db311 .update(crawlJobs)312 .set({ status: "queued", workerId: null })313 .where(and(eq(crawlJobs.status, "running"), lt(crawlJobs.heartbeatAt, new Date(Date.now() - STALE_MS))))314 .catch(() => {});315 while (!stopped) {316 try {317 if (activeJobs.size < MAX_PARALLEL_JOBS) {318 const claimed = await claimJob();319 if (claimed) {320 const p = runJob(claimed, log)321 .catch((e) => log.warn(`crawl ${claimed.id} crashed: ${(e as Error).message}`))322 .finally(() => activeJobs.delete(claimed.id));323 activeJobs.set(claimed.id, p);324 continue;325 }326 }327 } catch (e) {328 log.warn(`crawl worker loop error: ${(e as Error).message}`);329 }330 await new Promise<void>((r) => {331 wake = r;332 setTimeout(r, 4000);333 });334 wake = null;335 }336 };337 void loop();338 log.info(`crawl worker started (${WORKER_ID}, parallel jobs=${MAX_PARALLEL_JOBS})`);339 return async () => {340 stopped = true;341 kick();342 await Promise.allSettled([...activeJobs.values()]);343 running = false;344 };345}346347async function claimJob(): Promise<CrawlJob | null> {348 const rows = await db.execute(sql`349 update crawl_jobs set status = 'running', worker_id = ${WORKER_ID}, started_at = coalesce(started_at, now()), heartbeat_at = now()350 where id = (select id from crawl_jobs where status = 'queued' order by created_at asc limit 1 for update skip locked)351 returning *352 `);353 const r = (rows as unknown as { rows: Record<string, unknown>[] }).rows?.[0];354 if (!r) return null;355 const [job] = await db.select().from(crawlJobs).where(eq(crawlJobs.id, r.id as string)).limit(1);356 return job ?? null;357}358359interface Frontier {360 url: string;361 depth: number;362 parent: string | null;363}364365async function runJob(job: CrawlJob, log: { info: (m: string) => void; warn: (m: string) => void }): Promise<void> {366 const opts = crawlCreateSchema.parse(job.options) as CrawlCreateInput;367 const { principalForCrawl } = await import("../auth");368 let principal: ApiPrincipal;369 try {370 principal = await principalForCrawl(job.projectId, job.apiKeyId);371 } catch (e) {372 await finishJob(job.id, "failed", { code: (e as FetchaError).code ?? "INTERNAL_ERROR", message: (e as Error).message });373 return;374 }375 const seedUrl = new URL(job.seedUrl);376 const inScope = scopeFilter(seedUrl, opts.same_domain, opts.allow_subdomains);377 const include = urlPatternMatcher(opts.include_patterns);378 const exclude = urlPatternMatcher(opts.exclude_patterns);379 const seen = new Set<string>();380 const frontier: Frontier[] = [];381 let discovered = 0;382 let fetched = 0;383 let ok = 0;384 let blocked = 0;385 let failed = 0;386 let bytes = 0;387 let costUsd = 0;388 let robots: RobotsRules | null = null;389 let delayMs = opts.delay_ms;390391 // Resume support: already-known pages of this job.392 const existing = await db.select({ url: crawlPages.url, status: crawlPages.status, depth: crawlPages.depth, parentUrl: crawlPages.parentUrl }).from(crawlPages).where(eq(crawlPages.jobId, job.id));393 for (const p of existing) {394 seen.add(p.url);395 discovered++;396 if (p.status === "pending") frontier.push({ url: p.url, depth: p.depth, parent: p.parentUrl });397 else {398 fetched++;399 if (p.status === "success") ok++;400 else if (p.status === "blocked") blocked++;401 else if (p.status === "failed") failed++;402 }403 }404 bytes = job.bytes;405 costUsd = job.costUsd;406407 const fetchPage = async (url: string, extra: Record<string, unknown> = {}): Promise<FetchHandlerOutput | FetchaError> => {408 try {409 return await handleFetch({410 principal,411 body: {412 url,413 format: opts.format === "html" ? "html" : opts.format,414 timeout: opts.timeout,415 country: opts.country,416 network: opts.network,417 browser: opts.browser,418 browser_fallback: opts.browser_fallback,419 headers: opts.headers,420 links: true,421 ...extra,422 },423 source: "crawl",424 clientIp: null,425 userAgent: `fetcha-crawl/${job.id}`,426 });427 } catch (e) {428 return e instanceof FetchaError ? e : new FetchaError("INTERNAL_ERROR", (e as Error).message);429 }430 };431432 // robots.txt + optional sitemap seeding433 if (opts.respect_robots) {434 const r = await fetchPage(`${seedUrl.origin}/robots.txt`, { format: "raw", links: false, browser: false, browser_fallback: false, retries: 1, timeout: Math.min(opts.timeout, 15_000) });435 if (!(r instanceof FetchaError)) {436 const b = r.body as FetchResponseBody;437 if (b.success && b.content && (b.content_type ?? "").includes("text/plain")) {438 robots = parseRobots(b.content, "fetchabot");439 if (robots.crawlDelayMs && robots.crawlDelayMs > delayMs) delayMs = Math.min(robots.crawlDelayMs, 10_000);440 }441 }442 }443 const enqueue = async (candidates: Array<{ url: string; depth: number; parent: string | null }>) => {444 const rows: Array<{ id: string; jobId: string; url: string; depth: number; parentUrl: string | null; status: string }> = [];445 for (const c of candidates) {446 if (discovered >= opts.max_pages * 3 || frontier.length + fetched >= opts.max_pages) break;447 const n = normalizeCrawlUrl(c.url);448 if (!n || seen.has(n)) continue;449 if (!inScope(n) || !looksLikePage(n)) continue;450 if (exclude && exclude(n)) continue;451 if (include && !include(n) && c.depth > 0) continue;452 if (robots && !robotsAllows(robots, n)) continue;453 seen.add(n);454 discovered++;455 frontier.push({ url: n, depth: c.depth, parent: c.parent });456 rows.push({ id: newId("cpg"), jobId: job.id, url: n, depth: c.depth, parentUrl: c.parent, status: "pending" });457 }458 if (rows.length) await db.insert(crawlPages).values(rows).onConflictDoNothing().catch(() => {});459 };460 if (opts.use_sitemap && fetched === 0) {461 const sitemapUrls = new Set<string>([`${seedUrl.origin}/sitemap.xml`]);462 for (const s of robots?.sitemaps ?? []) sitemapUrls.add(s);463 let count = 0;464 for (const su of sitemapUrls) {465 if (count++ >= 5 || frontier.length >= opts.max_pages) break;466 const r = await fetchPage(su, { format: "raw", links: false, browser: false, browser_fallback: false, retries: 1 });467 if (r instanceof FetchaError) continue;468 const b = r.body as FetchResponseBody;469 if (!b.success || !b.content) continue;470 const { urls, sitemaps } = parseSitemap(b.content);471 for (const s of sitemaps.slice(0, 5)) sitemapUrls.add(s);472 await enqueue(urls.slice(0, opts.max_pages).map((u) => ({ url: u, depth: 1, parent: su })));473 }474 }475476 let lastHeartbeat = 0;477 let lastPersist = 0;478 const persistStats = async (force = false) => {479 const now = Date.now();480 if (!force && now - lastPersist < 2000) return;481 lastPersist = now;482 await db483 .update(crawlJobs)484 .set({ pagesDiscovered: discovered, pagesFetched: fetched, pagesOk: ok, pagesBlocked: blocked, pagesFailed: failed, bytes, costUsd, heartbeatAt: new Date() })485 .where(eq(crawlJobs.id, job.id))486 .catch(() => {});487 lastHeartbeat = now;488 };489 const isCancelled = async (): Promise<boolean> => {490 if (cancelled.has(job.id)) return true;491 if (Date.now() - lastHeartbeat > HEARTBEAT_MS) {492 const [row] = await db.select({ status: crawlJobs.status }).from(crawlJobs).where(eq(crawlJobs.id, job.id)).limit(1);493 lastHeartbeat = Date.now();494 if (row?.status === "cancelled") {495 cancelled.add(job.id);496 return true;497 }498 }499 return false;500 };501502 const worker = async () => {503 while (!stopped) {504 if (fetched >= opts.max_pages) return;505 if (await isCancelled()) return;506 const item = frontier.shift();507 if (!item) return;508 fetched++;509 const t0 = Date.now();510 const r = await fetchPage(item.url);511 const durationMs = Date.now() - t0;512 const update: Partial<typeof crawlPages.$inferInsert> = { fetchedAt: new Date(), durationMs };513 if (r instanceof FetchaError) {514 failed++;515 Object.assign(update, { status: "failed", errorCode: r.code, requestId: r.requestId ?? null });516 } else {517 const b = r.body as FetchResponseBody;518 const content = opts.format === "markdown" ? b.markdown : opts.format === "text" ? b.text : b.content;519 const isBlocked = !b.success && b.status >= 400 && r.summary && b.metadata.attempts > 0 && !(b.status === 404 || b.status === 410);520 bytes += r.summary?.bytes ?? b.metadata.bytes;521 costUsd += r.summary?.costUsd ?? 0;522 if (b.success) ok++;523 else if (isBlocked && b.status !== 500) blocked++;524 else failed++;525 Object.assign(update, {526 status: b.success ? "success" : isBlocked && b.status !== 500 ? "blocked" : "failed",527 httpStatus: b.status,528 errorCode: b.success ? null : b.status === 403 || b.status === 429 || b.status === 503 ? "TARGET_BLOCKED" : `HTTP_${b.status}`,529 requestId: b.request_id,530 finalUrl: b.final_url,531 title: b.page?.title ?? null,532 description: b.page?.description ?? null,533 contentType: b.content_type,534 content: b.success && typeof content === "string" ? content.slice(0, PAGE_CONTENT_CAP) : null,535 linksCount: b.page?.links_count ?? 0,536 bytes: r.summary?.bytes ?? b.metadata.bytes,537 mode: b.metadata.mode,538 });539 if (b.success && item.depth < opts.max_depth && b.links?.length) {540 await enqueue(b.links.filter((l) => l.internal || !opts.same_domain).map((l) => ({ url: l.url, depth: item.depth + 1, parent: item.url })));541 }542 }543 await db.update(crawlPages).set(update).where(and(eq(crawlPages.jobId, job.id), eq(crawlPages.url, item.url))).catch(() => {});544 await persistStats();545 if (delayMs > 0) await new Promise((res) => setTimeout(res, delayMs));546 }547 };548549 try {550 await Promise.all(Array.from({ length: Math.max(1, Math.min(opts.concurrency, 10)) }, () => worker()));551 await persistStats(true);552 if (cancelled.has(job.id)) {553 await finishJob(job.id, "cancelled");554 cancelled.delete(job.id);555 } else if (stopped && frontier.length) {556 await db.update(crawlJobs).set({ status: "queued", workerId: null }).where(eq(crawlJobs.id, job.id)).catch(() => {});557 } else {558 await finishJob(job.id, "completed");559 }560 log.info(`crawl ${job.id} ${cancelled.has(job.id) ? "cancelled" : "done"}: ${fetched} fetched, ${ok} ok, ${blocked} blocked, ${failed} failed`);561 if (opts.webhook_url && !stopped) await deliverWebhook(job.id, opts.webhook_url).catch(() => {});562 } catch (e) {563 await persistStats(true);564 await finishJob(job.id, "failed", { code: "INTERNAL_ERROR", message: (e as Error).message?.slice(0, 500) });565 throw e;566 }567}568569async function finishJob(id: string, status: "completed" | "failed" | "cancelled", error?: { code: string; message?: string }) {570 await db571 .update(crawlJobs)572 .set({ status, completedAt: new Date(), errorCode: error?.code ?? null, errorMessage: error?.message ?? null })573 .where(and(eq(crawlJobs.id, id), inArray(crawlJobs.status, ["running", "queued", "cancelled"])))574 .catch(() => {});575}576577async function deliverWebhook(jobId: string, url: string) {578 const [job] = await db.select().from(crawlJobs).where(eq(crawlJobs.id, jobId)).limit(1);579 if (!job) return;580 let status = "failed";581 try {582 await assertUrlAllowed(url);583 const ac = new AbortController();584 const t = setTimeout(() => ac.abort(), 10_000);585 const res = await fetch(url, { method: "POST", headers: { "content-type": "application/json", "user-agent": "fetcha-webhook/0.2" }, body: JSON.stringify({ type: "crawl.completed", job: serializeJob(job) }), signal: ac.signal });586 clearTimeout(t);587 status = res.ok ? `delivered:${res.status}` : `failed:${res.status}`;588 } catch (e) {589 status = `failed:${(e as Error).message?.slice(0, 80)}`;590 }591 await db.update(crawlJobs).set({ webhookStatus: status }).where(eq(crawlJobs.id, jobId)).catch(() => {});592}593