import {
  FetchaError,
  PLAN_LIMITS,
  assertUrlAllowed,
  crawlCreateSchema,
  extractDomain,
  extractPageMetadata,
  looksLikePage,
  mapCreateSchema,
  newId,
  normalizeCrawlUrl,
  parseRobots,
  parseSitemap,
  registrableHost,
  robotsAllows,
  urlPatternMatcher,
  type CrawlCreateInput,
  type FetchResponseBody,
  type MapCreateInput,
  type RobotsRules,
} from "@fetcha/core";
import { and, crawlJobs, crawlPages, db, desc, eq, inArray, lt, sql, type CrawlJob, type CrawlPage } from "@fetcha/db";
import { hostname } from "node:os";
import type { ApiPrincipal } from "../auth";
import { handleFetch, type FetchHandlerOutput } from "../routes/fetch";

/**
 * Crawl service: durable jobs (crawl_jobs / crawl_pages) executed in-process by a small worker
 * loop. Every page goes through `handleFetch`, so quotas, routing intelligence, block detection and
 * browser escalation apply exactly as for a single fetch.
 */

const WORKER_ID = `${hostname()}:${process.pid}`;
const PAGE_CONTENT_CAP = 600_000; // chars stored per page
const HEARTBEAT_MS = 10_000;
const STALE_MS = 3 * 60_000;

export function serializeJob(j: CrawlJob) {
  return {
    id: j.id,
    status: j.status,
    label: j.label,
    seed_url: j.seedUrl,
    domain: j.domain,
    source: j.source,
    options: j.options,
    stats: { discovered: j.pagesDiscovered, fetched: j.pagesFetched, ok: j.pagesOk, blocked: j.pagesBlocked, failed: j.pagesFailed, bytes: j.bytes },
    error: j.errorCode ? { code: j.errorCode, message: j.errorMessage } : null,
    webhook_status: j.webhookStatus,
    created_at: j.createdAt.toISOString(),
    started_at: j.startedAt?.toISOString() ?? null,
    completed_at: j.completedAt?.toISOString() ?? null,
  };
}

export function serializePage(p: CrawlPage, includeContent = true) {
  return {
    id: p.id,
    url: p.url,
    final_url: p.finalUrl,
    depth: p.depth,
    parent_url: p.parentUrl,
    status: p.status,
    http_status: p.httpStatus,
    error_code: p.errorCode,
    request_id: p.requestId,
    title: p.title,
    description: p.description,
    content_type: p.contentType,
    content: includeContent ? p.content : undefined,
    links_count: p.linksCount,
    bytes: p.bytes,
    duration_ms: p.durationMs,
    mode: p.mode,
    fetched_at: p.fetchedAt?.toISOString() ?? null,
  };
}

// ---------------------------------------------------------------------------
// Public operations
// ---------------------------------------------------------------------------
export async function createCrawl(principal: ApiPrincipal, body: unknown, source: "api" | "playground" | "sdk") {
  const parsed = crawlCreateSchema.safeParse(body ?? {});
  if (!parsed.success) {
    throw new FetchaError("INVALID_REQUEST", "Invalid crawl options.", { details: { issues: parsed.error.issues.map((i) => ({ path: i.path.join("."), message: i.message })) } });
  }
  const opts = parsed.data;
  const limits = PLAN_LIMITS[principal.plan];
  if (opts.max_pages > limits.crawl_max_pages) opts.max_pages = limits.crawl_max_pages;
  if (opts.timeout > limits.max_timeout_ms) opts.timeout = limits.max_timeout_ms;
  if (opts.network !== "auto" && !limits.networks.includes(opts.network)) throw new FetchaError("NETWORK_UNAVAILABLE");
  const allowed = await assertUrlAllowed(opts.url);
  const seed = normalizeCrawlUrl(allowed.url.toString());
  if (!seed) throw new FetchaError("INVALID_REQUEST", "The seed URL is invalid.");
  if (opts.webhook_url) await assertUrlAllowed(opts.webhook_url);

  const [active] = await db
    .select({ n: sql<number>`count(*)::int` })
    .from(crawlJobs)
    .where(and(eq(crawlJobs.organizationId, principal.organizationId), inArray(crawlJobs.status, ["queued", "running"])));
  if ((active?.n ?? 0) >= limits.crawl_concurrent_jobs) {
    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 } });
  }

  const id = newId("crawl");
  await db.insert(crawlJobs).values({
    id,
    organizationId: principal.organizationId,
    projectId: principal.projectId,
    apiKeyId: principal.keyId,
    source,
    label: opts.label ?? null,
    seedUrl: seed,
    domain: extractDomain(seed),
    options: { ...opts, url: seed } as Record<string, unknown>,
    status: "queued",
    pagesDiscovered: 1,
  });
  await db.insert(crawlPages).values({ id: newId("cpg"), jobId: id, url: seed, depth: 0, status: "pending" }).onConflictDoNothing();
  const [job] = await db.select().from(crawlJobs).where(eq(crawlJobs.id, id)).limit(1);
  kick();
  return serializeJob(job!);
}

export async function getCrawl(principal: ApiPrincipal, id: string) {
  const [job] = await db.select().from(crawlJobs).where(and(eq(crawlJobs.id, id), eq(crawlJobs.projectId, principal.projectId))).limit(1);
  if (!job) throw new FetchaError("CRAWL_NOT_FOUND");
  return serializeJob(job);
}

export async function listCrawls(principal: ApiPrincipal, limit = 50) {
  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));
  return { data: rows.map(serializeJob) };
}

export async function listCrawlPages(principal: ApiPrincipal, id: string, q: { cursor?: string; limit?: number; status?: string; include_content?: boolean }) {
  const [job] = await db.select({ id: crawlJobs.id }).from(crawlJobs).where(and(eq(crawlJobs.id, id), eq(crawlJobs.projectId, principal.projectId))).limit(1);
  if (!job) throw new FetchaError("CRAWL_NOT_FOUND");
  const limit = Math.min(Math.max(1, q.limit ?? 100), 500);
  const conds = [eq(crawlPages.jobId, id)];
  if (q.status) conds.push(eq(crawlPages.status, q.status));
  else conds.push(sql`${crawlPages.status} <> 'pending'`);
  if (q.cursor) {
    const [ts, cid] = decodeCursor(q.cursor);
    conds.push(sql`(${crawlPages.createdAt}, ${crawlPages.id}) > (${new Date(ts)}, ${cid})`);
  }
  const rows = await db
    .select()
    .from(crawlPages)
    .where(and(...conds))
    .orderBy(crawlPages.createdAt, crawlPages.id)
    .limit(limit + 1);
  const page = rows.slice(0, limit);
  const last = page.at(-1);
  return { data: page.map((p) => serializePage(p, q.include_content !== false)), next_cursor: rows.length > limit && last ? encodeCursor(last.createdAt, last.id) : null };
}

export async function cancelCrawl(principal: ApiPrincipal, id: string) {
  const res = await db
    .update(crawlJobs)
    .set({ status: "cancelled", completedAt: new Date() })
    .where(and(eq(crawlJobs.id, id), eq(crawlJobs.projectId, principal.projectId), inArray(crawlJobs.status, ["queued", "running"])))
    .returning({ id: crawlJobs.id });
  if (!res.length) {
    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);
    if (!job) throw new FetchaError("CRAWL_NOT_FOUND");
    return { id, status: job.status };
  }
  cancelled.add(id);
  return { id, status: "cancelled" };
}

function encodeCursor(d: Date, id: string): string {
  return Buffer.from(`${d.toISOString()}|${id}`).toString("base64url");
}
function decodeCursor(c: string): [string, string] {
  try {
    const [ts, id] = Buffer.from(c, "base64url").toString("utf8").split("|");
    if (!ts || !id || Number.isNaN(Date.parse(ts))) throw new Error("bad");
    return [ts, id];
  } catch {
    throw new FetchaError("INVALID_REQUEST", "Invalid cursor.");
  }
}

// ---------------------------------------------------------------------------
// Map (sync)
// ---------------------------------------------------------------------------
export async function mapSite(principal: ApiPrincipal, body: unknown, source: "api" | "playground" | "sdk", clientIp: string | null) {
  const parsed = mapCreateSchema.safeParse(body ?? {});
  if (!parsed.success) {
    throw new FetchaError("INVALID_REQUEST", "Invalid map options.", { details: { issues: parsed.error.issues.map((i) => ({ path: i.path.join("."), message: i.message })) } });
  }
  const opts = parsed.data;
  const allowed = await assertUrlAllowed(opts.url);
  const seed = normalizeCrawlUrl(allowed.url.toString())!;
  const seedUrl = new URL(seed);
  const host = registrableHost(seedUrl.hostname);
  const inScope = scopeFilter(seedUrl, opts.same_domain, opts.allow_subdomains);
  const search = opts.search ? urlPatternMatcher([opts.search.includes("*") || /^\/.+\/[a-z]*$/.test(opts.search) ? opts.search : `*${opts.search}*`]) : null;
  const deadline = Date.now() + Math.min(55_000, opts.timeout * 2);
  const found = new Map<string, "sitemap" | "links">();
  const add = (u: string, src: "sitemap" | "links") => {
    const n = normalizeCrawlUrl(u);
    if (!n || found.has(n) || !inScope(n) || !looksLikePage(n)) return;
    if (search && !search(n)) return;
    found.set(n, src);
  };
  const fetchText = async (url: string, format: "raw" | "html" = "raw", extra: Record<string, unknown> = {}): Promise<FetchResponseBody | null> => {
    if (Date.now() > deadline - 2000) return null;
    try {
      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}` });
      return out.body as FetchResponseBody;
    } catch {
      return null;
    }
  };
  let sitemapCount = 0;
  let linksCount = 0;
  if (opts.use_sitemap) {
    const sitemapUrls = new Set<string>([`${seedUrl.origin}/sitemap.xml`, `${seedUrl.origin}/sitemap_index.xml`]);
    const robots = await fetchText(`${seedUrl.origin}/robots.txt`);
    if (robots?.success && robots.content) for (const s of parseRobots(robots.content).sitemaps) sitemapUrls.add(s);
    const queue = [...sitemapUrls];
    const seen = new Set<string>();
    while (queue.length && seen.size < 25 && found.size < opts.limit && Date.now() < deadline - 3000) {
      const u = queue.shift()!;
      if (seen.has(u)) continue;
      seen.add(u);
      const r = await fetchText(u);
      if (!r?.success || !r.content || !/<(urlset|sitemapindex)|<loc>/i.test(r.content.slice(0, 5000)) && !/^https?:\/\//m.test(r.content.slice(0, 500))) continue;
      const { urls, sitemaps } = parseSitemap(r.content);
      for (const s of sitemaps) if (host === safeHost(s) || opts.allow_subdomains) queue.push(s);
      for (const x of urls) {
        if (found.size >= opts.limit) break;
        const before = found.size;
        add(x, "sitemap");
        if (found.size > before) sitemapCount++;
      }
    }
  }
  if (opts.use_links && found.size < opts.limit && Date.now() < deadline - 3000) {
    const r = await fetchText(seed, "html", { links: true });
    if (r?.success && r.links) {
      for (const l of r.links) {
        if (found.size >= opts.limit) break;
        const before = found.size;
        add(l.url, "links");
        if (found.size > before) linksCount++;
      }
    } else if (r?.content) {
      const meta = extractPageMetadata(r.content, r.final_url);
      for (const l of meta.links) {
        if (found.size >= opts.limit) break;
        const before = found.size;
        add(l.url, "links");
        if (found.size > before) linksCount++;
      }
    }
  }
  const urls = [...found.keys()];
  return { url: seed, count: urls.length, urls, sources: { sitemap: sitemapCount, links: linksCount }, truncated: urls.length >= opts.limit };
}

function safeHost(u: string): string {
  try {
    return registrableHost(new URL(u).hostname);
  } catch {
    return "";
  }
}

function scopeFilter(seed: URL, sameDomain: boolean, allowSubdomains: boolean): (url: string) => boolean {
  const seedHost = seed.hostname.toLowerCase().replace(/^www\./, "");
  const reg = registrableHost(seed.hostname);
  return (url: string) => {
    if (!sameDomain) return true;
    let h: string;
    try {
      h = new URL(url).hostname.toLowerCase().replace(/^www\./, "");
    } catch {
      return false;
    }
    if (h === seedHost) return true;
    if (allowSubdomains) return registrableHost(h) === reg;
    return false;
  };
}

// ---------------------------------------------------------------------------
// Worker
// ---------------------------------------------------------------------------
const cancelled = new Set<string>();
let running = false;
let stopped = false;
let wake: (() => void) | null = null;
const activeJobs = new Map<string, Promise<void>>();
const MAX_PARALLEL_JOBS = Number(process.env.FETCHA_CRAWL_PARALLEL_JOBS ?? 4);

function kick() {
  wake?.();
}

export function startCrawlWorker(log: { info: (m: string) => void; warn: (m: string) => void }): () => Promise<void> {
  if (running) return async () => {};
  running = true;
  stopped = false;
  const loop = async () => {
    // Re-queue jobs left "running" by a previous process (crash / redeploy).
    await db
      .update(crawlJobs)
      .set({ status: "queued", workerId: null })
      .where(and(eq(crawlJobs.status, "running"), lt(crawlJobs.heartbeatAt, new Date(Date.now() - STALE_MS))))
      .catch(() => {});
    while (!stopped) {
      try {
        if (activeJobs.size < MAX_PARALLEL_JOBS) {
          const claimed = await claimJob();
          if (claimed) {
            const p = runJob(claimed, log)
              .catch((e) => log.warn(`crawl ${claimed.id} crashed: ${(e as Error).message}`))
              .finally(() => activeJobs.delete(claimed.id));
            activeJobs.set(claimed.id, p);
            continue;
          }
        }
      } catch (e) {
        log.warn(`crawl worker loop error: ${(e as Error).message}`);
      }
      await new Promise<void>((r) => {
        wake = r;
        setTimeout(r, 4000);
      });
      wake = null;
    }
  };
  void loop();
  log.info(`crawl worker started (${WORKER_ID}, parallel jobs=${MAX_PARALLEL_JOBS})`);
  return async () => {
    stopped = true;
    kick();
    await Promise.allSettled([...activeJobs.values()]);
    running = false;
  };
}

async function claimJob(): Promise<CrawlJob | null> {
  const rows = await db.execute(sql`
    update crawl_jobs set status = 'running', worker_id = ${WORKER_ID}, started_at = coalesce(started_at, now()), heartbeat_at = now()
    where id = (select id from crawl_jobs where status = 'queued' order by created_at asc limit 1 for update skip locked)
    returning *
  `);
  const r = (rows as unknown as { rows: Record<string, unknown>[] }).rows?.[0];
  if (!r) return null;
  const [job] = await db.select().from(crawlJobs).where(eq(crawlJobs.id, r.id as string)).limit(1);
  return job ?? null;
}

interface Frontier {
  url: string;
  depth: number;
  parent: string | null;
}

async function runJob(job: CrawlJob, log: { info: (m: string) => void; warn: (m: string) => void }): Promise<void> {
  const opts = crawlCreateSchema.parse(job.options) as CrawlCreateInput;
  const { principalForCrawl } = await import("../auth");
  let principal: ApiPrincipal;
  try {
    principal = await principalForCrawl(job.projectId, job.apiKeyId);
  } catch (e) {
    await finishJob(job.id, "failed", { code: (e as FetchaError).code ?? "INTERNAL_ERROR", message: (e as Error).message });
    return;
  }
  const seedUrl = new URL(job.seedUrl);
  const inScope = scopeFilter(seedUrl, opts.same_domain, opts.allow_subdomains);
  const include = urlPatternMatcher(opts.include_patterns);
  const exclude = urlPatternMatcher(opts.exclude_patterns);
  const seen = new Set<string>();
  const frontier: Frontier[] = [];
  let discovered = 0;
  let fetched = 0;
  let ok = 0;
  let blocked = 0;
  let failed = 0;
  let bytes = 0;
  let costUsd = 0;
  let robots: RobotsRules | null = null;
  let delayMs = opts.delay_ms;

  // Resume support: already-known pages of this job.
  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));
  for (const p of existing) {
    seen.add(p.url);
    discovered++;
    if (p.status === "pending") frontier.push({ url: p.url, depth: p.depth, parent: p.parentUrl });
    else {
      fetched++;
      if (p.status === "success") ok++;
      else if (p.status === "blocked") blocked++;
      else if (p.status === "failed") failed++;
    }
  }
  bytes = job.bytes;
  costUsd = job.costUsd;

  const fetchPage = async (url: string, extra: Record<string, unknown> = {}): Promise<FetchHandlerOutput | FetchaError> => {
    try {
      return await handleFetch({
        principal,
        body: {
          url,
          format: opts.format === "html" ? "html" : opts.format,
          timeout: opts.timeout,
          country: opts.country,
          network: opts.network,
          browser: opts.browser,
          browser_fallback: opts.browser_fallback,
          headers: opts.headers,
          links: true,
          ...extra,
        },
        source: "crawl",
        clientIp: null,
        userAgent: `fetcha-crawl/${job.id}`,
      });
    } catch (e) {
      return e instanceof FetchaError ? e : new FetchaError("INTERNAL_ERROR", (e as Error).message);
    }
  };

  // robots.txt + optional sitemap seeding
  if (opts.respect_robots) {
    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) });
    if (!(r instanceof FetchaError)) {
      const b = r.body as FetchResponseBody;
      if (b.success && b.content && (b.content_type ?? "").includes("text/plain")) {
        robots = parseRobots(b.content, "fetchabot");
        if (robots.crawlDelayMs && robots.crawlDelayMs > delayMs) delayMs = Math.min(robots.crawlDelayMs, 10_000);
      }
    }
  }
  const enqueue = async (candidates: Array<{ url: string; depth: number; parent: string | null }>) => {
    const rows: Array<{ id: string; jobId: string; url: string; depth: number; parentUrl: string | null; status: string }> = [];
    for (const c of candidates) {
      if (discovered >= opts.max_pages * 3 || frontier.length + fetched >= opts.max_pages) break;
      const n = normalizeCrawlUrl(c.url);
      if (!n || seen.has(n)) continue;
      if (!inScope(n) || !looksLikePage(n)) continue;
      if (exclude && exclude(n)) continue;
      if (include && !include(n) && c.depth > 0) continue;
      if (robots && !robotsAllows(robots, n)) continue;
      seen.add(n);
      discovered++;
      frontier.push({ url: n, depth: c.depth, parent: c.parent });
      rows.push({ id: newId("cpg"), jobId: job.id, url: n, depth: c.depth, parentUrl: c.parent, status: "pending" });
    }
    if (rows.length) await db.insert(crawlPages).values(rows).onConflictDoNothing().catch(() => {});
  };
  if (opts.use_sitemap && fetched === 0) {
    const sitemapUrls = new Set<string>([`${seedUrl.origin}/sitemap.xml`]);
    for (const s of robots?.sitemaps ?? []) sitemapUrls.add(s);
    let count = 0;
    for (const su of sitemapUrls) {
      if (count++ >= 5 || frontier.length >= opts.max_pages) break;
      const r = await fetchPage(su, { format: "raw", links: false, browser: false, browser_fallback: false, retries: 1 });
      if (r instanceof FetchaError) continue;
      const b = r.body as FetchResponseBody;
      if (!b.success || !b.content) continue;
      const { urls, sitemaps } = parseSitemap(b.content);
      for (const s of sitemaps.slice(0, 5)) sitemapUrls.add(s);
      await enqueue(urls.slice(0, opts.max_pages).map((u) => ({ url: u, depth: 1, parent: su })));
    }
  }

  let lastHeartbeat = 0;
  let lastPersist = 0;
  const persistStats = async (force = false) => {
    const now = Date.now();
    if (!force && now - lastPersist < 2000) return;
    lastPersist = now;
    await db
      .update(crawlJobs)
      .set({ pagesDiscovered: discovered, pagesFetched: fetched, pagesOk: ok, pagesBlocked: blocked, pagesFailed: failed, bytes, costUsd, heartbeatAt: new Date() })
      .where(eq(crawlJobs.id, job.id))
      .catch(() => {});
    lastHeartbeat = now;
  };
  const isCancelled = async (): Promise<boolean> => {
    if (cancelled.has(job.id)) return true;
    if (Date.now() - lastHeartbeat > HEARTBEAT_MS) {
      const [row] = await db.select({ status: crawlJobs.status }).from(crawlJobs).where(eq(crawlJobs.id, job.id)).limit(1);
      lastHeartbeat = Date.now();
      if (row?.status === "cancelled") {
        cancelled.add(job.id);
        return true;
      }
    }
    return false;
  };

  const worker = async () => {
    while (!stopped) {
      if (fetched >= opts.max_pages) return;
      if (await isCancelled()) return;
      const item = frontier.shift();
      if (!item) return;
      fetched++;
      const t0 = Date.now();
      const r = await fetchPage(item.url);
      const durationMs = Date.now() - t0;
      const update: Partial<typeof crawlPages.$inferInsert> = { fetchedAt: new Date(), durationMs };
      if (r instanceof FetchaError) {
        failed++;
        Object.assign(update, { status: "failed", errorCode: r.code, requestId: r.requestId ?? null });
      } else {
        const b = r.body as FetchResponseBody;
        const content = opts.format === "markdown" ? b.markdown : opts.format === "text" ? b.text : b.content;
        const isBlocked = !b.success && b.status >= 400 && r.summary && b.metadata.attempts > 0 && !(b.status === 404 || b.status === 410);
        bytes += r.summary?.bytes ?? b.metadata.bytes;
        costUsd += r.summary?.costUsd ?? 0;
        if (b.success) ok++;
        else if (isBlocked && b.status !== 500) blocked++;
        else failed++;
        Object.assign(update, {
          status: b.success ? "success" : isBlocked && b.status !== 500 ? "blocked" : "failed",
          httpStatus: b.status,
          errorCode: b.success ? null : b.status === 403 || b.status === 429 || b.status === 503 ? "TARGET_BLOCKED" : `HTTP_${b.status}`,
          requestId: b.request_id,
          finalUrl: b.final_url,
          title: b.page?.title ?? null,
          description: b.page?.description ?? null,
          contentType: b.content_type,
          content: b.success && typeof content === "string" ? content.slice(0, PAGE_CONTENT_CAP) : null,
          linksCount: b.page?.links_count ?? 0,
          bytes: r.summary?.bytes ?? b.metadata.bytes,
          mode: b.metadata.mode,
        });
        if (b.success && item.depth < opts.max_depth && b.links?.length) {
          await enqueue(b.links.filter((l) => l.internal || !opts.same_domain).map((l) => ({ url: l.url, depth: item.depth + 1, parent: item.url })));
        }
      }
      await db.update(crawlPages).set(update).where(and(eq(crawlPages.jobId, job.id), eq(crawlPages.url, item.url))).catch(() => {});
      await persistStats();
      if (delayMs > 0) await new Promise((res) => setTimeout(res, delayMs));
    }
  };

  try {
    await Promise.all(Array.from({ length: Math.max(1, Math.min(opts.concurrency, 10)) }, () => worker()));
    await persistStats(true);
    if (cancelled.has(job.id)) {
      await finishJob(job.id, "cancelled");
      cancelled.delete(job.id);
    } else if (stopped && frontier.length) {
      await db.update(crawlJobs).set({ status: "queued", workerId: null }).where(eq(crawlJobs.id, job.id)).catch(() => {});
    } else {
      await finishJob(job.id, "completed");
    }
    log.info(`crawl ${job.id} ${cancelled.has(job.id) ? "cancelled" : "done"}: ${fetched} fetched, ${ok} ok, ${blocked} blocked, ${failed} failed`);
    if (opts.webhook_url && !stopped) await deliverWebhook(job.id, opts.webhook_url).catch(() => {});
  } catch (e) {
    await persistStats(true);
    await finishJob(job.id, "failed", { code: "INTERNAL_ERROR", message: (e as Error).message?.slice(0, 500) });
    throw e;
  }
}

async function finishJob(id: string, status: "completed" | "failed" | "cancelled", error?: { code: string; message?: string }) {
  await db
    .update(crawlJobs)
    .set({ status, completedAt: new Date(), errorCode: error?.code ?? null, errorMessage: error?.message ?? null })
    .where(and(eq(crawlJobs.id, id), inArray(crawlJobs.status, ["running", "queued", "cancelled"])))
    .catch(() => {});
}

async function deliverWebhook(jobId: string, url: string) {
  const [job] = await db.select().from(crawlJobs).where(eq(crawlJobs.id, jobId)).limit(1);
  if (!job) return;
  let status = "failed";
  try {
    await assertUrlAllowed(url);
    const ac = new AbortController();
    const t = setTimeout(() => ac.abort(), 10_000);
    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 });
    clearTimeout(t);
    status = res.ok ? `delivered:${res.status}` : `failed:${res.status}`;
  } catch (e) {
    status = `failed:${(e as Error).message?.slice(0, 80)}`;
  }
  await db.update(crawlJobs).set({ webhookStatus: status }).where(eq(crawlJobs.id, jobId)).catch(() => {});
}