feat(ingest): crawl discovery source (#586)

.changeset/crawl-discovery.md

@@ -0,0 +1,6 @@
+---
+'@chatcops/ingest': minor
+'@chatcops/cli': minor
+---
+
+Support crawl ingestion sources starting from a URL and following same-origin links breadth-first with maxDepth, maxPages, deduplication, robots.txt compliance, and cached HTML reuse during extraction. Add --crawl to the chatcops ingest command and crawl: URLs to --source.

api/ingest.api.md

@@ -27,6 +27,7 @@ export declare function discover(sources: IngestSource[], options: DiscoverOptio
 export interface DiscoveredPage {
     url: string;
     file?: string;
+    html?: string;
 }
 
 export interface DiscoverOptions {
@@ -35,6 +36,8 @@ export interface DiscoverOptions {
     timeoutMs: number;
     include?: string[];
     exclude?: string[];
+    concurrency?: number;
+    respectRobots?: boolean;
 }
 
 export interface Discovery {
@@ -114,6 +117,11 @@ export type IngestSource = {
     baseUrl: string;
     path: string;
     type: 'directory';
+} | {
+    maxDepth?: number;
+    maxPages?: number;
+    type: 'crawl';
+    url: string;
 };
 
 export interface IngestStats {

packages/cli/src/main.ts

@@ -34,8 +34,9 @@ Usage:
   chatcops test [files...] [options]        Run eval files against your bot
 
 Ingest options:
-  --source <url|path>   A sitemap (.xml, .xml.gz), a page URL, or a folder of .md/.mdx/.html
-                        files (with --base-url). Repeatable. Replaces the config's sources.
+  --source <url|path>   A sitemap (.xml, .xml.gz), crawl:<url>, a page URL, or a folder of
+                        .md/.mdx/.html files (with --base-url). Repeatable. Replaces the config's sources.
+  --crawl <url>         Crawl same-origin pages starting at url. Replaces the config's sources.
   --base-url <url>      Where a folder source is published
   --index <name>        Index name (default "default")
   --dry-run             Discover, extract and chunk; write and embed nothing
@@ -125,6 +126,7 @@ function parse(argv: string[]) {
     options: {
       config: { type: 'string' },
       source: { type: 'string', multiple: true },
+      crawl: { type: 'string' },
       'base-url': { type: 'string' },
       index: { type: 'string' },
       'dry-run': { type: 'boolean' },
@@ -158,7 +160,10 @@ async function runIngest(values: Values, io: Io): Promise<number> {
   const loaded = await loadConfig(io.cwd, values.config);
   try {
     const target = indexConfig(loaded);
-    const sources = values.source?.length ? values.source.map((source) => toSource(source, values['base-url'], io.cwd)) : loaded.config.ingest?.sources;
+    const cliSources: IngestSource[] = [];
+    if (values.crawl) cliSources.push({ type: 'crawl', url: values.crawl });
+    if (values.source?.length) cliSources.push(...values.source.map((source) => toSource(source, values['base-url'], io.cwd)));
+    const sources = cliSources.length ? cliSources : loaded.config.ingest?.sources;
     if (!sources?.length) throw new ConfigError('Nothing to ingest: pass --source, or set ingest.sources in the config.');
 
     // A model only runs with --contextualize, so a run never spends on one by surprise: the config holds which.
@@ -221,6 +226,9 @@ async function runSearch(question: string, values: Values, io: Io): Promise<numb
 
 /** A `--source` value: a sitemap URL, a page URL, or a folder (with `--base-url`). */
 export function toSource(value: string, baseUrl: string | undefined, cwd: string): IngestSource {
+  if (/^crawl:https?:\/\//i.test(value)) {
+    return { type: 'crawl', url: value.slice('crawl:'.length) };
+  }
   if (/^https?:\/\//i.test(value)) {
     return /\.xml(\.gz)?(\?.*)?$/i.test(value) ? { type: 'sitemap', url: value } : { type: 'urls', urls: [value] };
   }

packages/cli/tests/cli.test.ts

@@ -250,10 +250,28 @@ export default {
     const cwd = project({ shared: 'folder' });
     expect(await run(['ingest', '--source', './docs'], cwd)).toMatchObject({ code: EXIT.unusable, stderr: expect.stringContaining('needs --base-url') });
   });
+
+  it('ingests with --crawl and crawl: source', async () => {
+    site.set('/crawl/cli/start', '<!doctype html><html><body><main><h1>Start</h1><p>Start with <a href="/crawl/cli/sub">Sub</a></p></main></body></html>');
+    site.set('/crawl/cli/sub', '<!doctype html><html><body><main><h1>Sub</h1><p>Sub content</p></main></body></html>');
+    const cwd = project({ shared: 'cli-crawl' });
+    try {
+      const crawlFlag = await run(['ingest', '--crawl', `${site.url}/crawl/cli/start`], cwd);
+      expect(crawlFlag.code).toBe(EXIT.ok);
+      expect(crawlFlag.stdout).toContain('Indexed 2 of 2 pages');
+
+      const crawlSource = await run(['ingest', '--source', `crawl:${site.url}/crawl/cli/start`, '--force'], cwd);
+      expect(crawlSource.code).toBe(EXIT.ok);
+      expect(crawlSource.stdout).toContain('Indexed 2 of 2 pages');
+    } finally {
+      for (const p of ['/crawl/cli/start', '/crawl/cli/sub']) site.remove(p);
+    }
+  });
 });
 
 describe('toSource', () => {
   it('tells sitemaps, pages and folders apart', () => {
+    expect(toSource('crawl:https://acme.test/start', undefined, '/x')).toEqual({ type: 'crawl', url: 'https://acme.test/start' });
     expect(toSource('https://acme.test/sitemap.xml', undefined, '/x')).toEqual({ type: 'sitemap', url: 'https://acme.test/sitemap.xml' });
     expect(toSource('https://acme.test/sitemap.xml.gz?v=2', undefined, '/x')).toEqual({ type: 'sitemap', url: 'https://acme.test/sitemap.xml.gz?v=2' });
     expect(toSource('https://acme.test/pricing', undefined, '/x')).toEqual({ type: 'urls', urls: ['https://acme.test/pricing'] });

packages/ingest/src/discover.ts

@@ -1,6 +1,7 @@
 import { readdir } from 'node:fs/promises';
 import { join, relative, sep } from 'node:path';
 import { gunzipSync } from 'node:zlib';
+import { robotsFor, type Robots } from './robots.js';
 import type { DiscoveredPage, IngestSource } from './types.js';
 
 /** Sitemap indexes can point at more indexes; this many levels is plenty. */
@@ -13,6 +14,10 @@ export interface DiscoverOptions {
   timeoutMs: number;
   include?: string[];
   exclude?: string[];
+  /** Pages fetched at once during crawl. Default 4. */
+  concurrency?: number;
+  /** Obey robots.txt during crawl. Default true. */
+  respectRobots?: boolean;
 }
 
 export interface Discovery {
@@ -35,6 +40,8 @@ export async function discover(sources: IngestSource[], options: DiscoverOptions
       for (const url of source.urls) add({ url });
     } else if (source.type === 'sitemap') {
       await readSitemap(source.url, options, 0, new Set(), add, errors);
+    } else if (source.type === 'crawl') {
+      await crawlSite(source, options, add, errors);
     } else {
       try {
         for (const page of await readDirectory(source.path, source.baseUrl)) add(page);
@@ -103,6 +110,190 @@ async function readDirectory(path: string, baseUrl: string): Promise<DiscoveredP
     .sort((a, b) => a.url.localeCompare(b.url));
 }
 
+const DEFAULT_CRAWL_DEPTH = 3;
+const DEFAULT_CRAWL_PAGES = 500;
+
+/**
+ * Crawl a site breadth-first from `url`, following same-origin `<a href>`
+ * links. `maxDepth` is link hops from the start (depth 0); the start counts
+ * toward `maxPages`. Other origins, `mailto:`, `tel:`, fragment-only links
+ * and non-HTML responses are skipped. Robots.txt is respected when
+ * `respectRobots` is not false. Already-fetched HTML is stored on each page
+ * so extraction doesn't fetch again.
+ */
+async function crawlSite(
+  source: Extract<IngestSource, { type: 'crawl' }>,
+  options: DiscoverOptions,
+  add: (page: DiscoveredPage) => void,
+  errors: string[]
+): Promise<void> {
+  const maxDepth = source.maxDepth ?? DEFAULT_CRAWL_DEPTH;
+  const maxPages = source.maxPages ?? DEFAULT_CRAWL_PAGES;
+  if (maxPages <= 0) return;
+
+  const startUrl = normalizeUrl(source.url);
+  if (!startUrl) {
+    errors.push(`${source.url}: not a valid URL`);
+    return;
+  }
+
+  // Apply existing include/exclude rules before fetching, including the starting URL.
+  if (!matches(startUrl, options.include, options.exclude)) {
+    return;
+  }
+
+  const origin = new URL(startUrl).origin;
+  const respectRobots = options.respectRobots !== false;
+  const concurrency = Math.max(1, options.concurrency ?? 4);
+
+  const policies = new Map<string, Promise<Robots>>();
+  async function getRobots(urlOrOrigin: string): Promise<Robots | undefined> {
+    if (!respectRobots) return undefined;
+    const fetchWithPolicy = options.fetch as typeof fetch & { policy?: (url: string) => Promise<Robots> };
+    if (typeof fetchWithPolicy.policy === 'function') {
+      return fetchWithPolicy.policy(urlOrOrigin);
+    }
+    const o = new URL(urlOrOrigin).origin;
+    let pending = policies.get(o);
+    if (!pending) {
+      pending = robotsFor(o, ((input, init) =>
+        options.fetch(String(input), {
+          ...init,
+          headers: { 'user-agent': options.userAgent, ...(init?.headers as Record<string, string>) },
+          signal: AbortSignal.timeout(options.timeoutMs),
+        })) as typeof fetch).then((rules) => {
+        if (rules.unavailable) errors.push(`${o}/robots.txt unavailable (${rules.unavailable})`);
+        return rules;
+      });
+      policies.set(o, pending);
+    }
+    return pending;
+  }
+
+  let robots: Robots | undefined;
+  if (respectRobots) {
+    try {
+      robots = await getRobots(startUrl);
+    } catch {
+      robots = undefined;
+    }
+    if (robots && !robots.allowed(startUrl)) {
+      return;
+    }
+  }
+
+  const seen = new Set<string>();
+  seen.add(startUrl);
+
+  let currentLevel = [startUrl];
+  let depth = 0;
+  let count = 0;
+
+  while (currentLevel.length > 0 && depth <= maxDepth && count < maxPages) {
+    const nextLevel: string[] = [];
+    let nextIndex = 0;
+    let inFlight = 0;
+    const waiters: Array<() => void> = [];
+    const notifyWaiters = () => {
+      while (waiters.length > 0) waiters.shift()?.();
+    };
+
+    const isLastDepth = depth >= maxDepth;
+
+    async function worker(): Promise<void> {
+      while (true) {
+        if (count >= maxPages) return;
+        if (nextIndex >= currentLevel.length) return;
+
+        if (count + inFlight >= maxPages) {
+          if (inFlight === 0) return;
+          await new Promise<void>((resolve) => waiters.push(resolve));
+          continue;
+        }
+
+        const url = currentLevel[nextIndex++];
+        inFlight++;
+
+        try {
+          if (robots && !robots.allowed(url)) continue;
+
+          let response: Response;
+          try {
+            response = await options.fetch(url, {
+              headers: { 'user-agent': options.userAgent, accept: 'text/html,application/xhtml+xml' },
+              signal: AbortSignal.timeout(options.timeoutMs),
+              redirect: 'follow',
+            });
+          } catch (error) {
+            errors.push(`${url}: ${message(error)}`);
+            continue;
+          }
+
+          if (!response.ok) {
+            errors.push(`${url}: HTTP ${response.status}`);
+            continue;
+          }
+
+          const type = response.headers.get('content-type') ?? '';
+          if (type && !/html/i.test(type)) continue;
+
+          const html = await response.text();
+          count++;
+          add({ url, html });
+
+          if (!isLastDepth) {
+            for (const href of extractLinks(html, url, origin)) {
+              const normalized = normalizeUrl(href);
+              if (!normalized || seen.has(normalized)) continue;
+              if (!matches(normalized, options.include, options.exclude)) continue;
+              if (robots && !robots.allowed(normalized)) continue;
+              seen.add(normalized);
+              nextLevel.push(normalized);
+            }
+          }
+        } finally {
+          inFlight--;
+          notifyWaiters();
+        }
+      }
+    }
+
+    const workerCount = Math.min(concurrency, currentLevel.length);
+    await Promise.all(Array.from({ length: workerCount }, worker));
+
+    currentLevel = nextLevel;
+    depth++;
+  }
+}
+
+/** Same-origin `<a href>` links on a page, resolved and filtered. */
+function extractLinks(html: string, baseUrl: string, origin: string): string[] {
+  const links: string[] = [];
+  const anchorRegex = /<a\s+[^>]*?(?<=\s)href\s*=\s*(?:"([^"]*)"|'([^']*)'|([^\s>]+))[^>]*>/gi;
+  for (const match of html.matchAll(anchorRegex)) {
+    const raw = match[1] ?? match[2] ?? match[3];
+    if (!raw) continue;
+    const clean = raw.replace(/&amp;/g, '&').trim();
+    if (!clean) continue;
+    // Skip mailto:, tel:, and other non-http(s) schemes.
+    if (/^(mailto|tel|javascript|data|blob):/i.test(clean)) continue;
+    // Skip fragment-only links.
+    if (clean.startsWith('#')) continue;
+
+    let resolved: URL;
+    try {
+      resolved = new URL(clean, baseUrl);
+    } catch {
+      continue;
+    }
+    if (resolved.protocol !== 'http:' && resolved.protocol !== 'https:') continue;
+    // Same origin only.
+    if (resolved.origin !== origin) continue;
+    links.push(resolved.href);
+  }
+  return links;
+}
+
 /** The URL without its fragment, or undefined when it isn't http(s). */
 export function normalizeUrl(url: string): string | undefined {
   try {

packages/ingest/src/ingest.ts

@@ -79,11 +79,14 @@ export async function ingest(config: IngestConfig): Promise<IngestStats> {
     return pending;
   }
 
-  const send: typeof fetch = async (input, init) => {
-    const url = String(input);
-    const rules = config.respectRobots === false ? undefined : await policy(url);
-    return scheduled(url, init, rules?.crawlDelay);
-  };
+  const send: typeof fetch = Object.assign(
+    async (input: RequestInfo | URL, init?: RequestInit) => {
+      const url = String(input);
+      const rules = config.respectRobots === false ? undefined : await policy(url);
+      return scheduled(url, init, rules?.crawlDelay);
+    },
+    { policy }
+  );
 
   if (embedder.dimensions !== store.dimensions) {
     throw new Error(
@@ -102,7 +105,15 @@ export async function ingest(config: IngestConfig): Promise<IngestStats> {
     );
   }
 
-  const found = await discover(config.sources, { fetch: send, userAgent, timeoutMs, include: config.include, exclude: config.exclude });
+  const found = await discover(config.sources, {
+    fetch: send,
+    userAgent,
+    timeoutMs,
+    include: config.include,
+    exclude: config.exclude,
+    concurrency: config.concurrency,
+    respectRobots: config.respectRobots,
+  });
   if (config.maxPages !== undefined && config.maxPages >= 0 && found.pages.length > config.maxPages) {
     found.pages = found.pages.slice(0, Math.floor(config.maxPages));
     // Pages left out aren't removed pages, so pruning a partial run isn't safe either.
@@ -135,6 +146,10 @@ export async function ingest(config: IngestConfig): Promise<IngestStats> {
       const doc = /\.html?$/i.test(page.file) ? extractHtml(source, page.url, selectorFor(page.url)) : extractMarkdown(source, page.url);
       return { kind: 'page', doc };
     }
+    if (page.html !== undefined) {
+      const doc = extractHtml(page.html, page.url, selectorFor(page.url));
+      return { kind: 'page', doc };
+    }
     if (config.respectRobots !== false) {
       const rules = await policy(page.url);
       if (rules.unavailable) return { kind: 'skip', reason: `robots.txt unavailable (${rules.unavailable})` };

packages/ingest/src/types.ts

@@ -6,7 +6,9 @@ export type IngestSource =
   /** Pages listed by hand. */
   | { type: 'urls'; urls: string[] }
   /** Local `.md`, `.mdx` and `.html` files, published under `baseUrl`. */
-  | { type: 'directory'; path: string; baseUrl: string };
+  | { type: 'directory'; path: string; baseUrl: string }
+  /** Crawl a site starting at `url`, following same-origin links. */
+  | { type: 'crawl'; url: string; maxDepth?: number; maxPages?: number };
 
 export interface IngestConfig {
   /** Names the index, so a new one can be built next to the old. Default `default`. */
@@ -67,6 +69,8 @@ export interface DiscoveredPage {
   url: string;
   /** For directory sources: the file to read instead of fetching. */
   file?: string;
+  /** For crawl sources: the HTML already fetched during discovery, so extraction doesn't fetch again. */
+  html?: string;
 }
 
 export interface ExtractedDoc {