openvibers/openvibe.sources Commits

A crawl source type: a whole site from its sitemaps and links, robots and noindex honoured, recrawl with history (contracts 0.224.0) (#128)

OP OpenVibers committed 16 days ago · cf819c3888 · 15 files, +1394 −91

Changes

ddiff --git a/README.md b/README.md
iindex 0b15646..f7d252f 100644
---- a/README.md
++++ b/README.md
@@@ -53,7 +53,7 @@ start and flushes on stop.
## The registry
--A source records: `key`, `name`, `type` (`rss|atom|sitemap|jsonld|api|manual`), `category`
++A source records: `key`, `name`, `type` (`rss|atom|sitemap|jsonld|api|crawl|manual`), `category`
(`news|blog|reviews|deals|coupons|trade|hosting|crowdfunding|games|housing|food|jobs`, read from `sources.source@1`; housing, food and jobs since contracts 0.210.0), `endpoints` (for `api`: a field mapping, no code per
provider), `auth` (`none|header|bearer|query` with the **name** of an environment variable —
names must start with `SOURCES_CRED_`, so an entry can never point at this service's own
@@@ -106,6 +106,7 @@ succeeded. Health status: `healthy | stale | failing | never_fetched | disabled
| `sitemap` (`sitemap@1`) | `<urlset>` (+ Google News extension), `<sitemapindex>` | index children are listed, not followed |
| `jsonld` (`jsonld@1`) | schema.org Product, Offer/AggregateOffer, Review, NewsArticle, BlogPosting (+ Article) | prices, ratings and availability exactly as the page states; missing = null; embedded review text not kept; a page without JSON-LD is a `parse_error` |
| `api` (`api@1`) | JSON or XML official APIs via a registry mapping | `items_path`, `items_shape` (`array` default, `map` for records keyed by id), `fields`, `extra`; epoch fields only where mapped as such |
++| `crawl` (`jsonld@1`, `page@1`) | a whole site from its seeds: its sitemaps and, if asked, its links (see below) | only pages that publish JSON-LD are kept, unless `require_structured: false` keeps a `page` item from the page's own search metadata |
| `manual` (`manual@1`) | items entered through the API with an evidence URL | no fetcher |
XML with DTD entity declarations is refused; the parser expands no entities (values are decoded
@@@ -170,6 +171,65 @@ plain feed source keeps it). The registry refuses a pattern that does not compil
and `extract` on an api source. `dealnews-coupon-codes` (category `coupons`) is the first: the code, the shop and the
stated end date of the DealNews deals that state a code, exactly as written.
++### Crawl sources (`server/crawl.js`, contracts 0.224.0)
++
++The owner, 2026-10-28: "aggregate all websites and select deals on all sites… a full comprehensive crawler system… the
++open source Google". A crawl source walks one site from its endpoints (the seeds). Its `crawl` settings:
++
++| Field | Default | What it does |
++|---|---|---|
++| `sitemaps` | `true` | read the site's sitemaps: robots.txt `Sitemap` lines, else `/sitemap.xml`; indexes and `.xml.gz` followed |
++| `follow_links` | `false` | also follow same-host links (not `rel=nofollow`, not from a `nofollow` page) |
++| `max_depth` | `2` | with `follow_links`, how many links from a seed or sitemap page |
++| `include`, `exclude` | none | regular expressions over the path and query (seeds are fetched whatever they say) |
++| `max_pages` | `5000` | the frontier's size for the site |
++| `pages_per_run` | `100` | requests in one run (sitemaps count) |
++| `recrawl_sec` | `86400` | when a page that changed is due again; an unchanged page waits up to 4× longer |
++| `require_structured` | `true` | keep only pages with schema.org JSON-LD |
++
++**The frontier** (`crawl_frontier`, migration 0006) holds every page and sitemap the crawl knows, with its state:
++
++- **Page states:** `new`, `kept`, `hub`, `unstructured`, `noindex`, `duplicate`, `not_html`, `too_large`, `excluded`,
++ `denied`, `missing`, `gone` and `failing`.
++- **What it records per page:** when the page is due, its validators, its body hash, and the fetch run its items were
++ last read in.
++- **Run order:** each run reads the due sitemaps first, then the due pages, oldest first.
++
++**Every request is a source request like any other:**
++- robots.txt on the URL and on every redirect hop;
++- per-host spacing with Crawl-delay and the source's `min_interval_ms`;
++- conditional GETs;
++- the address guard;
++- one `fetch_runs` row per page.
++
++**Scope:** a crawl never leaves its seeds' hosts (or their `www` twins).
++
++**Pages that ask not to be kept:**
++- **Canonical:** a page whose canonical is another page of the crawl is read there instead.
++- **noindex:** a page that says noindex (`<meta name="robots">`, `<meta name="openvibesources">`, `X-Robots-Tag`) is
++ not kept, and its items are retired.
++- **Gone:** a page that answers 404 twice, or 410 once, is gone and its items are retired as `gone_from_source`.
++
++**A page's failure is the page's:** the source fails (backoff, one `sources.fetch.failed`) only when the site does:
++- robots.txt unreachable;
++- a 429 or a 503 with Retry-After;
++- five requests in a row with no usable answer.
++
++**Scheduling:**
++- at most `SOURCES_MAX_CONCURRENT_CRAWLS` (2) crawls run at once;
++- a run lasts at most `SOURCES_CRAWL_RUN_MS` (15 min);
++- while pages are still due, the next run comes after `SOURCES_CRAWL_SOON_MS` (60 s), not `poll_interval_sec`.
++
++**Settings that don't apply:** a crawl source has no `retire_after_sec`, because a page leaves with its items as
++described above.
++
++**API:**
++- `GET /api/v1/sources/:key/crawl` shows the frontier: pages by state, sitemaps, how many are due, and the latest
++ fetches.
++- `POST /api/v1/sources/:key/fetch` starts a crawl and answers 202.
++
++Tests: `test/crawl.test.js`.
++
### Items
Each item: `source_key`, `category`, `kind`, `identity`, `canonical_url` (tracking parameters and
ddiff --git a/docs/contracts-proposal/contracts/sources/source.v1.json b/docs/contracts-proposal/contracts/sources/source.v1.json
iindex 2165312..45e08bf 100644
---- a/docs/contracts-proposal/contracts/sources/source.v1.json
++++ b/docs/contracts-proposal/contracts/sources/source.v1.json
@@@ -38,6 +38,7 @@
"sitemap",
"jsonld",
"api",
++ "crawl",
"manual"
]
},
@@@ -151,7 +152,9 @@
"pattern": "^[a-z][a-z0-9_]{1,39}$"
},
"request_content_type": {
-- "enum": ["application/json"],
++ "enum": [
++ "application/json"
++ ],
"description": "Sent as the request's Content-Type on this endpoint's GET (no body). Open Collective's GraphQL API answers a GET only with it, as its CSRF guard (Sources#65)."
},
"max_bytes": {
@@@ -249,6 +252,72 @@
}
}
},
++ "crawl": {
++ "type": [
++ "object",
++ "null"
++ ],
++ "description": "type crawl only: how the crawler walks one site from its endpoints (the seeds). Every request still passes the site's robots.txt (RFC 9309) and the per-host spacing, a page that says noindex is not kept, and only what a page publishes for machines is read (schema.org JSON-LD: products and offers for deals, articles, reviews). A crawl adds pages to its frontier from the site's own sitemaps and, if asked, its links; each run fetches the pages that are due and schedules them again after recrawl_sec, so a price seen twice becomes a history.",
++ "additionalProperties": false,
++ "properties": {
++ "sitemaps": {
++ "type": "boolean",
++ "description": "Find pages from the site's sitemaps: the Sitemap lines of its robots.txt, else /sitemap.xml, following sitemap indexes (default true)."
++ },
++ "follow_links": {
++ "type": "boolean",
++ "description": "Also follow links on crawled pages to pages of the same host that match include (default false)."
++ },
++ "include": {
++ "type": "array",
++ "maxItems": 20,
++ "items": {
++ "type": "string",
++ "minLength": 1,
++ "maxLength": 200
++ },
++ "description": "Regular expressions over a URL's path and query; a page must match one (none given: every page of the host)."
++ },
++ "exclude": {
++ "type": "array",
++ "maxItems": 20,
++ "items": {
++ "type": "string",
++ "minLength": 1,
++ "maxLength": 200
++ },
++ "description": "Regular expressions over a URL's path and query; a page matching one is never fetched."
++ },
++ "max_pages": {
++ "type": "integer",
++ "minimum": 1,
++ "maximum": 200000,
++ "description": "The most pages the frontier holds for this site (default 5000)."
++ },
++ "pages_per_run": {
++ "type": "integer",
++ "minimum": 1,
++ "maximum": 1000,
++ "description": "Pages fetched in one run (default 100); the host spacing still applies between them."
++ },
++ "recrawl_sec": {
++ "type": "integer",
++ "minimum": 3600,
++ "maximum": 2592000,
++ "description": "How long after a fetch a page is due again (default 86400, a day)."
++ },
++ "max_depth": {
++ "type": "integer",
++ "minimum": 0,
++ "maximum": 10,
++ "description": "With follow_links, how many links away from a seed or sitemap page a page may be (default 2)."
++ },
++ "require_structured": {
++ "type": "boolean",
++ "description": "Keep only pages that publish schema.org JSON-LD the adapter reads (default true); a page without it is fetched once and dropped from the frontier."
++ }
++ }
++ },
"auth": {
"type": "object",
"required": [
@@@ -314,7 +383,10 @@
"minimum": 1
},
"retire_after_sec": {
-- "type": ["integer", "null"],
++ "type": [
++ "integer",
++ "null"
++ ],
"minimum": 3600,
"maximum": 31536000,
"description": "Retire what the feed stopped showing (jobs, listings): once a fetch of this source succeeds, an item it has not seen for this many seconds is removed with reason \"gone_from_source\" (a change, so consumers following the change feed learn it). null (the default): items stay, as a news archive or a dataset should."
ddiff --git a/migrations/0006_crawl.sql b/migrations/0006_crawl.sql
nnew file mode 100644
iindex 0000000..3d61ab4
---- /dev/null
++++ b/migrations/0006_crawl.sql
@@@ -0,0 +1,31 @@
++-- phase: expand
++-- A crawl source (sources.source@1 type crawl, openvibe-contracts 0.224.0; the owner, 2026-10-28: aggregate every site, "the
++-- open source Google") walks one site from its seeds. Its settings are the source's crawl column (JSON, null for every other
++-- type); the pages and sitemaps it knows are its frontier (server/crawl.js). robots.txt's Sitemap lines are kept with its
++-- rules, so a crawl finds a site's sitemaps without asking for robots.txt twice.
++ALTER TABLE sources ADD COLUMN IF NOT EXISTS crawl text COLLATE "C";
++
++ALTER TABLE robots_cache ADD COLUMN IF NOT EXISTS sitemaps text COLLATE "C" NOT NULL DEFAULT '[]';
++
++CREATE TABLE IF NOT EXISTS crawl_frontier (
++ source_key text COLLATE "C" NOT NULL,
++ url text COLLATE "C" NOT NULL,
++ kind text COLLATE "C" NOT NULL, -- page | sitemap
++ depth bigint NOT NULL DEFAULT 0, -- links away from a seed or sitemap page
++ found_via text COLLATE "C" NOT NULL, -- seed | robots | guess | sitemap | link | canonical
++ state text COLLATE "C" NOT NULL DEFAULT 'new',
++ first_seen_at bigint NOT NULL,
++ next_fetch_at bigint, -- null: not fetched again
++ last_fetch_at bigint,
++ interval_sec bigint, -- grows while the page stays the same, back to recrawl_sec when it changes
++ lastmod_at bigint, -- the sitemap's lastmod for it
++ last_status bigint,
++ etag text COLLATE "C",
++ last_modified text COLLATE "C",
++ body_hash text COLLATE "C",
++ last_ok_run_id text COLLATE "C", -- the fetch run its items were last read in
++ items bigint NOT NULL DEFAULT 0,
++ fails bigint NOT NULL DEFAULT 0,
++ PRIMARY KEY (source_key, url)
++);
++CREATE INDEX IF NOT EXISTS idx_crawl_frontier_due ON crawl_frontier (source_key, kind, next_fetch_at);
ddiff --git a/package-lock.json b/package-lock.json
iindex 5c880c5..0fec7c1 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -12,7 +12,7 @@
"dotenv": "^16.4.7",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.222.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.224.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.67.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.26.1",
"pg": "^8.23.0"
@@@ -769,9 +769,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.222.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.222.0",
-- "integrity": "sha512-wdUQCilq9glRMwfxEswbIj3l0QQqKDXZAi2+CR5Mlh34rZjI1Q+7GlAoR0UtNwZAgtLwXhPxFyJ/aPk8lcDG1Q==",
++ "version": "0.224.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.224.0",
++ "integrity": "sha512-/bLQ7ToMvOM9IACyOxygDtUwkMXhnPo6m954zPGOee1en3DwMFprdGOuRY/DMoPI1XVsTU0vyzfd+l6U7YIArg==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex da04ddd..faffd75 100644
---- a/package.json
++++ b/package.json
@@@ -27,7 +27,7 @@
"dotenv": "^16.4.7",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.222.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.224.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.67.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.26.1",
"pg": "^8.23.0"
ddiff --git a/server/adapters/page.js b/server/adapters/page.js
nnew file mode 100644
iindex 0000000..b922148
---- /dev/null
++++ b/server/adapters/page.js
@@@ -0,0 +1,138 @@
++'use strict';
++/**
++ * What a crawled HTML page says about itself for machines: its robots directives (<meta name="robots"> and our own
++ * product token's <meta>, plus X-Robots-Tag), its canonical URL, its links, and the metadata it publishes for search
++ * engines and link previews (title, description, Open Graph, article times, language).
++ *
++ * A crawl reads the page's schema.org JSON-LD first (adapters/jsonld.js); a source that does not require structured
++ * data keeps a page without it as one `page` item built from this metadata. Nothing here reads the page's prose: the
++ * summary is the description the page wrote for search results, never its body.
++ */
++const { toText, toIsoDate, canonicalUrl, decodeEntities } = require('./util');
++
++const PARSER_VERSION = 'page@1';
++const MAX_LINKS = 1000;
++
++/** The attributes of one start tag, names lower-cased, values decoded. */
++function attrs(tag) {
++ const out = {};
++ const re = /([^\s"'<>\/=]+)(?:\s*=\s*(?:"([^"]*)"|'([^']*)'|([^\s"'=<>`]+)))?/g;
++ const body = tag.replace(/^<\s*[a-z0-9-]+/i, '').replace(/\/?>$/, '');
++ let m;
++ while ((m = re.exec(body))) {
++ const name = m[1].toLowerCase();
++ if (!(name in out)) out[name] = decodeEntities(m[2] ?? m[3] ?? m[4] ?? '');
++ }
++ return out;
++}
++
++function tags(html, name) {
++ const re = new RegExp(`<${name}\\b[^>]*>`, 'gi');
++ return html.match(re) || [];
++}
++
++/** noindex / nofollow from a robots directive list ("noindex, nofollow", "none"). */
++function directives(value) {
++ const d = String(value || '').toLowerCase().split(',').map(s => s.trim());
++ return { noindex: d.includes('noindex') || d.includes('none'), nofollow: d.includes('nofollow') || d.includes('none') };
++}
++
++/**
++ * X-Robots-Tag: "noindex", "noindex, nofollow", or scoped to one crawler ("otherbot: noindex"). A directive scoped to
++ * another crawler does not apply to us; one scoped to our product token does.
++ */
++function headerDirectives(header, agent) {
++ const out = { noindex: false, nofollow: false };
++ for (const line of [].concat(header || [])) {
++ let value = String(line);
++ const m = /^\s*([a-z0-9_-]+)\s*:\s*(.*)$/i.exec(value);
++ if (m && !['unavailable_after', 'max-snippet', 'max-image-preview', 'max-video-preview'].includes(m[1].toLowerCase())) {
++ if (m[1].toLowerCase() !== String(agent).toLowerCase()) continue;
++ value = m[2];
++ }
++ const d = directives(value);
++ out.noindex = out.noindex || d.noindex;
++ out.nofollow = out.nofollow || d.nofollow;
++ }
++ return out;
++}
++
++/**
++ * read(html, { url, agent, headers, links }) → { noindex, nofollow, canonical, base, title, description, image, site_name,
++ * og_type, language, published_at, updated_at, author, links: [absolute urls] }
++ */
++function read(html, { url, agent = 'OpenVibeSources', headers = {}, links = false } = {}) {
++ const head = html.slice(0, 512 * 1024);
++ const meta = {};
++ let noindex = false;
++ let nofollow = false;
++ const token = String(agent).toLowerCase();
++ for (const t of tags(head, 'meta')) {
++ const a = attrs(t);
++ const key = (a.name || a.property || '').toLowerCase();
++ if (!key || a.content == null) continue;
++ if (key === 'robots' || key === token) {
++ const d = directives(a.content);
++ noindex = noindex || d.noindex;
++ nofollow = nofollow || d.nofollow;
++ continue;
++ }
++ if (!(key in meta)) meta[key] = a.content;
++ }
++ const h = headerDirectives(headers['x-robots-tag'], agent);
++ noindex = noindex || h.noindex;
++ nofollow = nofollow || h.nofollow;
++
++ let base = url;
++ const baseTag = tags(head, 'base').map(attrs).find(a => a.href);
++ if (baseTag) base = canonicalUrl(baseTag.href, url) || url;
++ let canonical = null;
++ for (const t of tags(head, 'link')) {
++ const a = attrs(t);
++ if (String(a.rel || '').toLowerCase().split(/\s+/).includes('canonical') && a.href) { canonical = canonicalUrl(a.href, base); break; }
++ }
++ const titleTag = /<title\b[^>]*>([\s\S]*?)<\/title\s*>/i.exec(head);
++ const htmlTag = tags(head, 'html')[0];
++ const lang = htmlTag ? attrs(htmlTag).lang : null;
++
++ const found = [];
++ if (links && !nofollow) {
++ const seen = new Set();
++ for (const t of tags(html, 'a')) {
++ if (found.length >= MAX_LINKS) break;
++ const a = attrs(t);
++ if (!a.href || String(a.rel || '').toLowerCase().split(/\s+/).includes('nofollow')) continue;
++ const abs = canonicalUrl(a.href, base);
++ if (abs && !seen.has(abs)) { seen.add(abs); found.push(abs); }
++ }
++ }
++
++ return {
++ noindex, nofollow, canonical, base,
++ title: toText(meta['og:title'] || (titleTag ? titleTag[1] : null), 500),
++ description: toText(meta['og:description'] || meta.description, 1000),
++ image: canonicalUrl(meta['og:image'] || meta['og:image:url'] || meta['twitter:image'], base),
++ site_name: toText(meta['og:site_name'], 200),
++ og_type: toText(meta['og:type'], 60),
++ language: lang ? String(lang).slice(0, 20) : null,
++ published_at: toIsoDate(meta['article:published_time']),
++ updated_at: toIsoDate(meta['article:modified_time'] || meta['og:updated_time']),
++ author: toText(meta.author || meta['article:author'], 200),
++ links: found,
++ };
++}
++
++/** The page as one item from its metadata (a source that does not require structured data); null without a title. */
++function pageItem(meta, pageUrl, summaryMax = 1000) {
++ if (!meta.title) return null;
++ const url = meta.canonical || pageUrl;
++ return {
++ identity: url, kind: 'page', canonical_url: url,
++ title: meta.title, summary: meta.description ? meta.description.slice(0, summaryMax) : null,
++ authors: meta.author && !/^https?:/i.test(meta.author) ? [meta.author] : [],
++ published_at: meta.published_at, updated_at: meta.updated_at,
++ fields: { site_name: meta.site_name, image: meta.image, language: meta.language, og_type: meta.og_type },
++ };
++}
++
++module.exports = { read, pageItem, attrs, headerDirectives, PARSER_VERSION };
ddiff --git a/server/api/sources.js b/server/api/sources.js
iindex bd761fa..bb9e50e 100644
---- a/server/api/sources.js
++++ b/server/api/sources.js
@@@ -6,13 +6,14 @@
* GET /api/v1/sources every source with health and staleness
* GET /api/v1/sources/:key
* GET /api/v1/sources/:key/runs?before=&limit= fetch runs, newest first
++ * GET /api/v1/sources/:key/crawl a crawl source's frontier: pages by state, due, the latest fetches
* GET /api/v1/runs?state=&source=&after=&limit= fetch runs across sources, oldest first
* GET /api/v1/health status counts and every source that is not healthy
* sources.source.manage
* POST /api/v1/sources create (disabled unless enabled: true + notes)
* PATCH /api/v1/sources/:key change fields (enable/disable too)
* DELETE /api/v1/sources/:key only while it has no items
-- * POST /api/v1/sources/:key/fetch run now (409 while a run is in flight)
++ * POST /api/v1/sources/:key/fetch run now (409 while a run is in flight); a crawl starts and answers 202
*/
const express = require('express');
const { http } = require('openvibe-contracts');
@@@ -66,6 +67,13 @@ function sourcesRouter({ db, registry, ingest, auth, limits }) {
res.json({ runs: rows.map(runView), next_before: rows.length === limit ? rows[rows.length - 1].rid : null });
});
++ router.get('/api/v1/sources/:key/crawl', read, async (req, res) => {
++ const out = await ingest.crawlStatus(req.params.key);
++ if (!out) return notFound(res, req);
++ if (out.not_crawl) return http.sendProblem(res, 409, 'sources.not_crawl', { detail: 'only a crawl source has a frontier', ctx: req.ov });
++ res.json({ crawl: out });
++ });
++
router.get('/api/v1/runs', read, async (req, res) => {
const state = req.query.state ? String(req.query.state) : null;
if (state && state !== 'failed' && !STATES.includes(state)) return http.sendProblem(res, 400, 'sources.bad_query', { detail: `state must be failed or one of ${STATES.join('|')}`, ctx: req.ov });
@@@ -116,6 +124,13 @@ function sourcesRouter({ db, registry, ingest, auth, limits }) {
router.post('/api/v1/sources/:key/fetch', fetchNow, async (req, res, next) => {
try {
++ const row = await registry.get(req.params.key);
++ if (row && row.type === 'crawl') {
++ // A crawl run takes minutes: it starts now and its pages show up in /runs and /crawl as they are fetched.
++ if (ingest.inflight().includes(row.key)) return http.sendProblem(res, 409, 'sources.fetch_in_flight', { detail: 'a run of this source is in flight', ctx: req.ov });
++ ingest.run(row.key, { trigger: 'manual' }).catch(err => console.error(`[sources] crawl ${row.key}: ${err.stack || err}`));
++ return res.status(202).json({ started: true, source_key: row.key });
++ }
const out = await ingest.run(req.params.key, { trigger: 'manual' });
if (!out) return notFound(res, req);
if (out.manual) return http.sendProblem(res, 409, 'sources.manual_source', { detail: 'a manual source is not fetched; add items with POST /api/v1/sources/:key/items', ctx: req.ov });
ddiff --git a/server/config.js b/server/config.js
iindex 7834546..6c0676d 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -81,6 +81,17 @@ function load(env = process.env) {
// deliberate on-host sources only). Empty in production.
allowPrivateHosts: list(env.SOURCES_ALLOW_PRIVATE_HOSTS, []),
},
++ // Crawl sources (server/crawl.js): how long one run may take, how many crawl at once (each holds a worker slot
++ // for its whole run, so feeds always keep the rest), sitemap limits, and how soon a run follows one that left
++ // pages due.
++ crawl: {
++ runMs: int(env.SOURCES_CRAWL_RUN_MS, 15 * 60 * 1000),
++ maxConcurrent: int(env.SOURCES_MAX_CONCURRENT_CRAWLS, 2),
++ soonMs: int(env.SOURCES_CRAWL_SOON_MS, 60 * 1000),
++ sitemapMaxBytes: int(env.SOURCES_SITEMAP_MAX_BYTES, 50 * 1024 * 1024),
++ sitemapMaxUrls: 50000,
++ maxSitemaps: 1000,
++ },
items: {
// The ceiling on any one source's max_items. Each source still keeps its own (most are 200 or less); 12000
// lets sec-company-tickers take SEC's whole ticker list (about 10,450 rows) in its one daily request.
ddiff --git a/server/crawl.js b/server/crawl.js
nnew file mode 100644
iindex 0000000..5b3b0c8
---- /dev/null
++++ b/server/crawl.js
@@@ -0,0 +1,498 @@
++'use strict';
++/**
++ * Crawl sources (sources.source@1 type crawl; the owner, 2026-10-28: "aggregate all websites… a full comprehensive
++ * crawler system… the open source Google"). A crawl walks one site from its endpoints (the seeds) and reads, page by
++ * page, only what the site publishes for machines: schema.org JSON-LD (products and offers for deals, articles,
++ * reviews) and, for a source that does not require it, the page's own search metadata (adapters/page.js).
++ *
++ * The frontier (crawl_frontier) is every page and sitemap the crawl knows, with when each is due again:
++ * - found from the seeds, the site's sitemaps (robots.txt Sitemap lines, else /sitemap.xml; indexes followed, and a
++ * newer <lastmod> makes a known page due again), its canonical links and, with follow_links, its same-host links
++ * up to max_depth; include/exclude patterns, robots.txt and max_pages decide what is added
++ * - each run fetches the sitemaps that are due, then up to pages_per_run due pages, oldest first
++ * - a page is due again recrawl_sec after a fetch that changed something; while it stays the same the wait grows to
++ * four times that, and a change brings it back: a price seen twice becomes a history (Deals' observations)
++ *
++ * Every request is a request like any other source's: robots.txt (RFC 9309) on the URL and every redirect hop, the
++ * per-host spacing with Crawl-delay and the source's min_interval_ms, conditional GETs, the address guard, a fetch_runs
++ * row per page. A crawl never leaves its seeds' hosts (or their www twins). A page that says noindex (meta robots, our
++ * own token's meta, X-Robots-Tag) is not kept and its items are retired; nofollow stops its links from being followed.
++ *
++ * Page states: new | kept (items read) | hub (no items, links followed) | unstructured (no items: dropped) | noindex |
++ * duplicate (its canonical is another page) | not_html | too_large | excluded | denied (robots) | missing (404 once) |
++ * gone (404/410 twice: its items are retired) | failing. Sitemaps: new | read | denied | missing | gone | failing.
++ *
++ * A failing page is the page's business; the source fails (backoff, sources.fetch.failed) only when the site does:
++ * robots.txt unreachable, a 429 or a 503 with Retry-After, or five requests in a row that got no usable answer.
++ */
++const zlib = require('zlib');
++const jsonld = require('./adapters/jsonld');
++const sitemap = require('./adapters/sitemap');
++const page = require('./adapters/page');
++const { sha256, decodeBody, canonicalUrl, ParseError } = require('./adapters/util');
++
++const HTML_ACCEPT = 'text/html, application/xhtml+xml;q=0.9, */*;q=0.1';
++const XML_ACCEPT = 'application/xml, text/xml;q=0.9, application/x-gzip;q=0.5, */*;q=0.1';
++const DAY_MS = 86400 * 1000;
++const GONE_RECHECK_MS = 30 * DAY_MS;
++const STREAK_LIMIT = 5;
++
++const looksLikeSitemap = (u) => /\.xml(\.gz)?$/i.test(new URL(u).pathname);
++const twins = (host) => (host.startsWith('www.') ? [host, host.slice(4)] : [host, `www.${host}`]);
++
++function retryAfterMs(value, t) {
++ if (value == null) return null;
++ const s = String(value).trim();
++ if (/^\d{1,7}$/.test(s)) return Number(s) * 1000;
++ const d = Date.parse(s);
++ return Number.isFinite(d) ? Math.max(0, d - t) : null;
++}
++
++function createCrawler({ db, config, items, robots, fetcher, spacer, record, now = () => Date.now(), log = console }) {
++ const st = {
++ seed: db.prepare(`INSERT INTO crawl_frontier (source_key, url, kind, depth, found_via, first_seen_at, next_fetch_at)
++ VALUES (@k, @url, @kind, 0, @via, @now, @now) ON CONFLICT (source_key, url) DO NOTHING`),
++ count: db.prepare('SELECT COUNT(*) AS n FROM crawl_frontier WHERE source_key = ? AND kind = ?'),
++ addMany: db.prepare(`INSERT INTO crawl_frontier (source_key, url, kind, depth, found_via, first_seen_at, next_fetch_at, lastmod_at)
++ SELECT @k, t.u, @kind, @depth, @via, @now, @now, t.lm FROM unnest(@urls::text[], @lms::bigint[]) AS t(u, lm)
++ WHERE NOT EXISTS (SELECT 1 FROM crawl_frontier f WHERE f.source_key = @k AND f.url = t.u)
++ LIMIT @room
++ ON CONFLICT (source_key, url) DO NOTHING`),
++ // A sitemap that says a known page changed after we last fetched it makes the page due now.
++ touchLastmod: db.prepare(`UPDATE crawl_frontier f SET lastmod_at = t.lm, next_fetch_at = LEAST(COALESCE(f.next_fetch_at, @now), @now)
++ FROM unnest(@urls::text[], @lms::bigint[]) AS t(u, lm)
++ WHERE f.source_key = @k AND f.url = t.u AND t.lm IS NOT NULL AND f.last_fetch_at IS NOT NULL AND t.lm > f.last_fetch_at
++ AND f.state NOT IN ('noindex', 'duplicate', 'excluded')`),
++ due: db.prepare(`SELECT * FROM crawl_frontier WHERE source_key = @k AND kind = @kind AND next_fetch_at <= @now
++ ORDER BY next_fetch_at, url LIMIT @limit`),
++ dueCount: db.prepare('SELECT COUNT(*) AS n FROM crawl_frontier WHERE source_key = ? AND next_fetch_at <= ?'),
++ stateOf: db.prepare('SELECT state FROM crawl_frontier WHERE source_key = ? AND url = ?'),
++ save: db.prepare(`UPDATE crawl_frontier SET state = @state, next_fetch_at = @next_fetch_at, last_fetch_at = @last_fetch_at,
++ interval_sec = @interval_sec, last_status = @last_status, etag = @etag, last_modified = @last_modified, body_hash = @body_hash,
++ last_ok_run_id = @last_ok_run_id, items = @items, fails = @fails WHERE source_key = @source_key AND url = @url`),
++ summary: db.prepare(`SELECT kind, state, COUNT(*) AS n, SUM(CASE WHEN next_fetch_at <= @now THEN 1 ELSE 0 END) AS due
++ FROM crawl_frontier WHERE source_key = @k GROUP BY kind, state`),
++ recent: db.prepare(`SELECT url, kind, state, depth, found_via, last_fetch_at, next_fetch_at, last_status, items, fails
++ FROM crawl_frontier WHERE source_key = ? AND last_fetch_at IS NOT NULL ORDER BY last_fetch_at DESC, url LIMIT ?`),
++ nextDue: db.prepare('SELECT MIN(next_fetch_at) AS at FROM crawl_frontier WHERE source_key = ?'),
++ };
++
++ const save = (r, changes) => st.save.run({ ...r, ...changes });
++
++ /** The crawl's reach: its hosts, its include/exclude patterns, and robots.txt per origin (one compiled matcher a run). */
++ function scopeOf(source) {
++ const c = source.crawl;
++ const hosts = new Set();
++ for (const e of source.endpoints) for (const h of twins(new URL(e.url).host)) hosts.add(h);
++ const include = c.include.map(p => new RegExp(p));
++ const exclude = c.exclude.map(p => new RegExp(p));
++ const matchers = new Map();
++ return {
++ inHost: (u) => hosts.has(u.host),
++ wanted(u) {
++ const pq = u.pathname + u.search;
++ if (exclude.some(re => re.test(pq))) return false;
++ return !include.length || include.some(re => re.test(pq));
++ },
++ async allowed(u) {
++ let m = matchers.get(u.origin);
++ if (!m) { m = await robots.matcher(u.href); matchers.set(u.origin, m); }
++ return m(u.pathname + u.search);
++ },
++ };
++ }
++
++ /** Add URLs to the frontier (in scope, wanted, allowed, within the cap); a newer lastmod makes a known page due. → added */
++ async function enqueue(source, scope, list, { kind, depth, via }) {
++ const urls = [];
++ const lms = [];
++ const seen = new Set();
++ for (const it of list) {
++ let u;
++ try { u = new URL(it.url); } catch { continue; }
++ if (seen.has(u.href) || !scope.inHost(u)) continue;
++ if (kind === 'page' && !scope.wanted(u)) continue;
++ if (!(await scope.allowed(u))) continue;
++ seen.add(u.href);
++ urls.push(u.href);
++ lms.push(Number.isFinite(it.lastmod) ? it.lastmod : null);
++ }
++ const cap = kind === 'sitemap' ? config.crawl.maxSitemaps : source.crawl.max_pages;
++ const t = now();
++ let added = 0;
++ for (let i = 0; i < urls.length; i += 1000) {
++ const chunk = { k: source.key, kind, depth, via, now: t, urls: urls.slice(i, i + 1000), lms: lms.slice(i, i + 1000) };
++ if (kind === 'page') await st.touchLastmod.run(chunk);
++ const room = cap - Number((await st.count.get(source.key, kind)).n);
++ if (room > 0) added += (await st.addMany.run({ ...chunk, room })).changes || 0;
++ }
++ return added;
++ }
++
++ /** Every run: the seeds, and the sitemaps robots.txt names (else /sitemap.xml). Known rows are left as they are. */
++ async function seed(source, scope) {
++ const t = now();
++ const origins = new Set();
++ for (const e of source.endpoints) {
++ const url = canonicalUrl(e.url);
++ if (!url) continue;
++ origins.add(new URL(url).origin);
++ await st.seed.run({ k: source.key, url, kind: looksLikeSitemap(url) ? 'sitemap' : 'page', via: 'seed', now: t });
++ }
++ if (!source.crawl.sitemaps) return;
++ for (const origin of origins) {
++ const named = await robots.sitemaps(origin);
++ const list = named.length ? named : [`${origin}/sitemap.xml`];
++ for (const s of list) {
++ const url = canonicalUrl(s);
++ if (url && scope.inHost(new URL(url))) await st.seed.run({ k: source.key, url, kind: 'sitemap', via: named.length ? 'robots' : 'guess', now: t });
++ }
++ }
++ }
++
++ /** A redirect stays on the crawl's hosts, passes robots.txt, and a new host is spaced like any request to it. */
++ const hop = (source, scope, host) => async (next) => {
++ if (!scope.inHost(next)) return { refuse: `${next.host} is outside this crawl` };
++ const v = await robots.check(next.toString());
++ if (!v.allowed) return { refuse: v.reason };
++ if (next.host !== host) {
++ spacer.setCrawlDelay(next.host, v.crawlDelaySec);
++ await spacer.space(next.host, source.min_interval_ms);
++ }
++ return null;
++ };
++
++ const backoffMs = (source, fails) => Math.min(source.crawl.recrawl_sec * 4000, 15 * 60 * 1000 * 2 ** Math.min(fails - 1, 10));
++ const grown = (source, prev) => Math.min(Math.round((prev || source.crawl.recrawl_sec) * 1.5), source.crawl.recrawl_sec * 4);
++
++ /**
++ * robots.txt for the URL, then the request. → { res } | { out } when it ended before or at the request (the run row is
++ * recorded and the frontier row saved).
++ */
++ async function request(source, scope, r, run, { accept, capBytes }) {
++ const u = new URL(r.url);
++ const verdict = await robots.check(r.url);
++ spacer.setCrawlDelay(u.host, verdict.crawlDelaySec);
++ if (!verdict.allowed) {
++ if (verdict.refusedCode || /^robots\.txt unreachable/.test(verdict.reason || '')) {
++ // The whole host is closed to us for now: the source's failure, not this page's.
++ Object.assign(run, verdict.refusedCode
++ ? { state: 'http_error', error_code: verdict.refusedCode, detail: `${u.host} is not fetchable (${verdict.refusedCode})` }
++ : { state: 'robots_denied', error_code: 'robots', detail: verdict.reason }, { finished_at: now() });
++ await record.insertRun(run);
++ return { out: { run, stop: true } };
++ }
++ Object.assign(run, { state: 'robots_denied', error_code: 'robots', detail: verdict.reason, finished_at: now() });
++ await record.insertRun(run);
++ await save(r, { state: 'denied', next_fetch_at: now() + config.fetch.robotsTtlMs });
++ return { out: { run } };
++ }
++ await spacer.space(u.host, source.min_interval_ms);
++ await record.requested(now(), source.key);
++ try {
++ const res = await fetcher.fetchUrl(r.url, {
++ headers: { Accept: accept }, etag: r.etag || null, lastModified: r.last_modified || null,
++ capBytes, beforeHop: hop(source, scope, u.host),
++ });
++ Object.assign(run, { http_status: res.status, bytes: res.bytes != null ? res.bytes : null, finished_at: now() });
++ return { res };
++ } catch (err) {
++ const t = now();
++ const state = err.code === 'timeout' ? 'timeout' : err.code === 'hop_refused' ? 'robots_denied' : 'http_error';
++ Object.assign(run, { state, error_code: err.code || 'network', detail: String(err.message).slice(0, 500), finished_at: t });
++ await record.insertRun(run);
++ if (['address_refused', 'port_refused'].includes(err.code)) return { out: { run, stop: true } };
++ if (err.code === 'hop_refused') {
++ await save(r, { state: 'denied', last_fetch_at: t, next_fetch_at: t + source.crawl.recrawl_sec * 1000 });
++ return { out: { run, requested: true, answered: true } };
++ }
++ if (err.code === 'too_large' || err.code === 'too_many_redirects' || err.code === 'bad_url') {
++ await save(r, { state: err.code === 'too_large' ? 'too_large' : 'failing', last_fetch_at: t, next_fetch_at: err.code === 'too_large' ? null : t + source.crawl.recrawl_sec * 4000 });
++ return { out: { run, requested: true, answered: true } };
++ }
++ const fails = Number(r.fails) + 1;
++ await save(r, { state: r.state === 'new' ? 'failing' : r.state, fails, last_fetch_at: t, next_fetch_at: t + backoffMs(source, fails) });
++ return { out: { run, requested: true, unanswered: true } };
++ }
++ }
++
++ /** Statuses a page and a sitemap answer alike; null when the body is for the caller to read. */
++ async function common(source, r, run, res) {
++ const t = run.finished_at;
++ if (res.status === 429 || (res.status === 503 && res.headers['retry-after'] != null)) {
++ const wait = Math.min(retryAfterMs(res.headers['retry-after'], t) ?? 10 * 60 * 1000, 24 * 3600 * 1000);
++ Object.assign(run, res.status === 429
++ ? { state: 'rate_limited', error_code: 'upstream_429', detail: `the site asked us to wait ${Math.round(wait / 1000)} s` }
++ : { state: 'http_error', error_code: 'http_503', detail: `HTTP 503, retry after ${Math.round(wait / 1000)} s` });
++ await record.insertRun(run);
++ await save(r, { next_fetch_at: t + wait });
++ return { run, requested: true, stop: true, notBefore: t + wait };
++ }
++ if (res.status === 404 || res.status === 410) {
++ const gone = res.status === 410 || [404, 410].includes(Number(r.last_status));
++ Object.assign(run, { state: 'http_error', error_code: `http_${res.status}`, detail: gone ? 'gone: its items are retired' : `HTTP ${res.status}, checked again tomorrow` });
++ await record.insertRun(run);
++ if (gone && r.last_ok_run_id) await items.retireRun(source, r.last_ok_run_id, 'gone_from_source');
++ await save(r, { state: gone ? 'gone' : 'missing', last_status: res.status, last_fetch_at: t, next_fetch_at: t + (gone ? GONE_RECHECK_MS : DAY_MS), etag: null, last_modified: null });
++ return { run, requested: true, answered: true };
++ }
++ if (res.status !== 200 && res.status !== 203 && res.status !== 304) {
++ Object.assign(run, { state: 'http_error', error_code: `http_${res.status}`, detail: `HTTP ${res.status}` });
++ await record.insertRun(run);
++ const fails = Number(r.fails) + 1;
++ await save(r, { state: r.state === 'new' ? 'failing' : r.state, fails, last_status: res.status, last_fetch_at: t, next_fetch_at: t + backoffMs(source, fails) });
++ // A server error or a refusal (401, 403) is the site not answering us; a 400 or 451 is about the page.
++ return { run, requested: true, ...(res.status >= 500 || res.status === 401 || res.status === 403 ? { unanswered: true } : { answered: true }) };
++ }
++ return null;
++ }
++
++ // ── A page ───────────────────────────────────────────────
++
++ async function fetchPage(source, scope, r, trigger) {
++ const c = source.crawl;
++ const u = new URL(r.url);
++ if (!scope.inHost(u) || (r.found_via !== 'seed' && !scope.wanted(u))) {
++ // The source's patterns changed since the page was found (a seed is fetched whatever they say): nothing is requested.
++ await save(r, { state: 'excluded', next_fetch_at: null });
++ return {};
++ }
++ const run = record.baseRun(source, r.url, trigger, now());
++ const req = await request(source, scope, r, run, { accept: HTML_ACCEPT });
++ if (req.out) return req.out;
++ const { res } = req;
++ const t = run.finished_at;
++ const early = await common(source, r, run, res);
++ if (early) return early;
++
++ const unchanged = async (hash) => {
++ const interval = grown(source, r.interval_sec);
++ Object.assign(run, { state: res.status === 304 ? 'not_modified' : 'ok', detail: res.status === 304 ? null : 'the page is the same' });
++ if (hash) run.raw_body_hash = hash;
++ await record.insertRun(run);
++ if (r.last_ok_run_id) await items.seenAgain(source, r.last_ok_run_id, t);
++ await save(r, {
++ last_status: res.status, last_fetch_at: t, interval_sec: interval, next_fetch_at: t + interval * 1000, fails: 0,
++ etag: res.headers.etag || r.etag, last_modified: res.headers['last-modified'] || r.last_modified,
++ });
++ return { run, requested: true, answered: true };
++ };
++ if (res.status === 304) return await unchanged(null);
++
++ const type = String(res.headers['content-type'] || '').toLowerCase();
++ if (type && !/html|xml/.test(type)) {
++ Object.assign(run, { state: 'ok', detail: `not a web page (${type.split(';')[0]}): not kept` });
++ await record.insertRun(run);
++ await save(r, { state: 'not_html', last_status: res.status, last_fetch_at: t, next_fetch_at: null });
++ return { run, requested: true, answered: true };
++ }
++ const hash = sha256(res.body);
++ run.raw_body_hash = hash;
++ if (hash === r.body_hash && ['kept', 'hub'].includes(r.state)) return await unchanged(hash);
++
++ const text = decodeBody(res.body, res.headers['content-type']);
++ const finalUrl = canonicalUrl(res.finalUrl) || r.url;
++ const follow = c.follow_links && Number(r.depth) < c.max_depth;
++ const meta = page.read(text, { url: finalUrl, agent: config.fetch.robotsAgent, headers: res.headers, links: follow });
++ const validators = { etag: res.headers.etag || null, last_modified: res.headers['last-modified'] || null };
++
++ if (meta.noindex) {
++ // The site asked search engines not to keep this page: we keep nothing from it either.
++ const retired = r.last_ok_run_id ? await items.retireRun(source, r.last_ok_run_id, 'noindex') : 0;
++ Object.assign(run, { state: 'ok', detail: `noindex: not kept${retired ? `, ${retired} item(s) retired` : ''}` });
++ await record.insertRun(run);
++ await save(r, { state: 'noindex', last_status: res.status, last_fetch_at: t, next_fetch_at: t + GONE_RECHECK_MS, body_hash: hash, items: 0, fails: 0, ...validators });
++ return { run, requested: true, answered: true };
++ }
++
++ // A page whose canonical is another page of the crawl is read there instead (unless that page points back here).
++ if (meta.canonical && meta.canonical !== r.url && meta.canonical !== finalUrl) {
++ const cu = new URL(meta.canonical);
++ if (scope.inHost(cu) && scope.wanted(cu)) {
++ const known = await st.stateOf.get(source.key, meta.canonical);
++ const added = known ? 0 : await enqueue(source, scope, [{ url: meta.canonical }], { kind: 'page', depth: Number(r.depth), via: 'canonical' });
++ if ((known && known.state !== 'duplicate') || added) {
++ Object.assign(run, { state: 'ok', detail: `canonical is ${meta.canonical}` });
++ await record.insertRun(run);
++ await save(r, { state: 'duplicate', last_status: res.status, last_fetch_at: t, next_fetch_at: null, body_hash: hash, fails: 0, ...validators });
++ return { run, requested: true, answered: true };
++ }
++ }
++ }
++
++ const linked = meta.links.length ? await enqueue(source, scope, meta.links.map(url => ({ url })), { kind: 'page', depth: Number(r.depth) + 1, via: 'link' }) : 0;
++
++ let parsed = { items: [], skipped: 0 };
++ let parser = jsonld.PARSER_VERSION;
++ try {
++ parsed = jsonld.parse(text, { url: finalUrl, summaryMax: config.items.summaryMax, maxItems: source.max_items });
++ } catch (err) {
++ if (!(err instanceof ParseError)) log.error(`[crawl] ${source.key}: jsonld crashed on ${r.url}: ${err.stack || err}`);
++ }
++ let found = parsed.items;
++ if (!found.length && !c.require_structured) {
++ const it = page.pageItem(meta, finalUrl, config.items.summaryMax);
++ if (it) { found = [it]; parser = page.PARSER_VERSION; }
++ }
++
++ if (!found.length) {
++ // A page that only leads to others stays a hub; one that never had anything is dropped, one that lost its data
++ // is looked at again later.
++ const hub = follow && meta.links.length > 0;
++ const state = hub ? 'hub' : 'unstructured';
++ const next = hub ? t + grown(source, r.interval_sec) * 1000 : r.state === 'kept' ? t + c.recrawl_sec * 4000 : null;
++ Object.assign(run, { state: 'ok', items_skipped: parsed.skipped, detail: hub ? `no structured data; ${linked} new link(s) followed` : 'no structured data: not kept' });
++ await record.insertRun(run);
++ await save(r, { state, last_status: res.status, last_fetch_at: t, next_fetch_at: next, interval_sec: hub ? grown(source, r.interval_sec) : r.interval_sec, body_hash: hash, items: 0, fails: 0, ...validators });
++ return { run, requested: true, answered: true };
++ }
++
++ // The only path that touches items: the run row, the items and the frontier row commit together.
++ await db.tx(async () => {
++ const counts = await items.ingest(source, found, { runId: run.id, rawBodyHash: hash, parserVersion: parser, retrievedAt: t });
++ Object.assign(run, {
++ state: 'ok', parser_version: parser, items_seen: found.length, items_created: counts.created, items_updated: counts.updated,
++ items_unchanged: counts.unchanged, items_skipped: parsed.skipped + counts.skipped, detail: linked ? `${linked} new link(s) followed` : null,
++ });
++ await record.insertRun(run);
++ const changed = counts.created + counts.updated > 0;
++ const interval = changed ? c.recrawl_sec : grown(source, r.interval_sec);
++ await save(r, {
++ state: 'kept', last_status: res.status, last_fetch_at: t, interval_sec: interval, next_fetch_at: t + interval * 1000,
++ body_hash: hash, last_ok_run_id: run.id, items: found.length, fails: 0, ...validators,
++ });
++ });
++ return { run, requested: true, answered: true };
++ }
++
++ // ── A sitemap ────────────────────────────────────────────
++
++ async function fetchSitemap(source, scope, r, trigger) {
++ const run = record.baseRun(source, r.url, trigger, now());
++ const req = await request(source, scope, r, run, { accept: XML_ACCEPT, capBytes: config.crawl.sitemapMaxBytes });
++ if (req.out) return req.out;
++ const { res } = req;
++ const t = run.finished_at;
++ const early = await common(source, r, run, res);
++ if (early) return early;
++ const again = t + Math.min(source.crawl.recrawl_sec * 1000, DAY_MS);
++ const validators = { etag: res.headers.etag || null, last_modified: res.headers['last-modified'] || null };
++ if (res.status === 304) {
++ Object.assign(run, { state: 'not_modified' });
++ await record.insertRun(run);
++ await save(r, { state: 'read', last_status: 304, last_fetch_at: t, next_fetch_at: again, fails: 0 });
++ return { run, requested: true, answered: true };
++ }
++ const hash = sha256(res.body);
++ run.raw_body_hash = hash;
++ run.parser_version = sitemap.PARSER_VERSION;
++ if (hash === r.body_hash && r.state === 'read') {
++ Object.assign(run, { state: 'ok', detail: 'the sitemap is the same' });
++ await record.insertRun(run);
++ await save(r, { last_status: res.status, last_fetch_at: t, next_fetch_at: again, fails: 0, ...validators });
++ return { run, requested: true, answered: true };
++ }
++ let parsed;
++ try {
++ let body = res.body;
++ // A .xml.gz sitemap is a gzip file, not a gzip-encoded response.
++ if (body.length > 2 && body[0] === 0x1f && body[1] === 0x8b) body = zlib.gunzipSync(body, { maxOutputLength: config.crawl.sitemapMaxBytes });
++ parsed = sitemap.parse(decodeBody(body, res.headers['content-type']), { url: res.finalUrl, maxItems: config.crawl.sitemapMaxUrls });
++ } catch (err) {
++ Object.assign(run, { state: 'parse_error', error_code: err instanceof ParseError ? 'unreadable' : 'adapter_error', detail: String(err.message).slice(0, 500) });
++ await record.insertRun(run);
++ const fails = Number(r.fails) + 1;
++ await save(r, { state: 'failing', fails, last_status: res.status, last_fetch_at: t, next_fetch_at: t + backoffMs(source, fails) });
++ return { run, requested: true, answered: true };
++ }
++ const index = parsed.meta.format === 'sitemapindex';
++ const list = parsed.items.map(it => ({ url: it.canonical_url, lastmod: it.updated_at ? Date.parse(it.updated_at) : null }));
++ const added = await enqueue(source, scope, list, index ? { kind: 'sitemap', depth: 0, via: 'sitemap' } : { kind: 'page', depth: 0, via: 'sitemap' });
++ Object.assign(run, { state: 'ok', items_seen: parsed.items.length, items_skipped: parsed.skipped, detail: `${added} new ${index ? 'sitemap' : 'page'}(s)` });
++ await record.insertRun(run);
++ await save(r, { state: 'read', last_status: res.status, last_fetch_at: t, next_fetch_at: again, body_hash: hash, items: parsed.items.length, fails: 0, ...validators });
++ return { run, requested: true, answered: true };
++ }
++
++ // ── One run ──────────────────────────────────────────────
++
++ /**
++ * run(source, { trigger, stopping }) → { runs, anyFailure, anySuccess, failedRun, notBefore, more }
++ * Sitemaps first, then pages, until pages_per_run requests, SOURCES_CRAWL_RUN_MS, a stop, or nothing due.
++ */
++ async function run(source, { trigger = 'schedule', stopping = () => false } = {}) {
++ const scope = scopeOf(source);
++ const out = { runs: [], anyFailure: false, anySuccess: false, failedRun: null, notBefore: 0, more: false };
++ const deadline = now() + config.crawl.runMs;
++ await seed(source, scope);
++ let budget = source.crawl.pages_per_run;
++ let streak = 0;
++ const done = new Set();
++ const live = () => budget > 0 && !stopping() && now() < deadline;
++ crawl: while (live()) {
++ let batch = (await st.due.all({ k: source.key, kind: 'sitemap', now: now(), limit: 5 })).filter(r => !done.has(r.url));
++ if (!batch.length) batch = (await st.due.all({ k: source.key, kind: 'page', now: now(), limit: Math.min(budget, 50) })).filter(r => !done.has(r.url));
++ if (!batch.length) break;
++ for (const r of batch) {
++ if (!live()) break crawl;
++ done.add(r.url);
++ let res;
++ try {
++ res = r.kind === 'sitemap' ? await fetchSitemap(source, scope, r, trigger) : await fetchPage(source, scope, r, trigger);
++ } catch (err) {
++ // Anything unexpected is a recorded failure of that page, never a silent gap or a stuck frontier row.
++ log.error(`[crawl] ${source.key}: ${r.url}: ${err.stack || err}`);
++ const failed = record.baseRun(source, r.url, trigger, now());
++ Object.assign(failed, { state: 'http_error', error_code: 'internal', detail: String(err.message).slice(0, 300) });
++ await record.insertRun(failed);
++ const fails = Number(r.fails) + 1;
++ await save(r, { fails, next_fetch_at: now() + backoffMs(source, fails) });
++ res = { run: failed, requested: true, unanswered: true };
++ }
++ if (res.run) out.runs.push(res.run);
++ if (res.requested) budget--;
++ if (res.answered) { out.anySuccess = true; streak = 0; }
++ if (res.unanswered) streak++;
++ if (res.stop || streak >= STREAK_LIMIT) {
++ out.anyFailure = true;
++ out.failedRun = res.run;
++ out.notBefore = res.notBefore || 0;
++ break crawl;
++ }
++ }
++ }
++ out.more = Number((await st.dueCount.get(source.key, now())).n) > 0;
++ // Nothing was due: the crawl is up to date, which is a success.
++ if (!out.anyFailure && !out.runs.length) out.anySuccess = true;
++ return out;
++ }
++
++ /** The frontier at a glance: counts by kind and state (with how many are due), the next due time, the latest fetches. */
++ async function status(source, { recent = 20 } = {}) {
++ const t = now();
++ const rows = await st.summary.all({ k: source.key, now: t });
++ const by = { page: {}, sitemap: {} };
++ let pages = 0;
++ let due = 0;
++ for (const row of rows) {
++ (by[row.kind] || (by[row.kind] = {}))[row.state] = Number(row.n);
++ if (row.kind === 'page') pages += Number(row.n);
++ due += Number(row.due || 0);
++ }
++ const next = (await st.nextDue.get(source.key)).at;
++ const iso = (v) => (v == null ? null : new Date(Number(v)).toISOString());
++ return {
++ pages, max_pages: source.crawl.max_pages, due, next_due_at: iso(next),
++ pages_by_state: by.page, sitemaps_by_state: by.sitemap,
++ recent: (await st.recent.all(source.key, recent)).map(row => ({
++ url: row.url, kind: row.kind, state: row.state, depth: Number(row.depth), found_via: row.found_via,
++ last_fetch_at: iso(row.last_fetch_at), next_fetch_at: iso(row.next_fetch_at), http_status: row.last_status == null ? null : Number(row.last_status),
++ items: Number(row.items), fails: Number(row.fails),
++ })),
++ };
++ }
++
++ return { run, status };
++}
++
++module.exports = { createCrawler, looksLikeSitemap };
ddiff --git a/server/ingest.js b/server/ingest.js
iindex e7943f0..29df1d2 100644
---- a/server/ingest.js
++++ b/server/ingest.js
@@@ -20,10 +20,15 @@
* A source with retire_after_sec (jobs, listings) retires what its feed stopped showing: after a run
* in which every endpoint answered (ok or 304) and none failed, its items not seen for that long are
* removed as gone_from_source. A 304 counts as seeing again what the endpoint's last good parse saw.
++ *
++ * A crawl source (type crawl) walks its site's frontier instead of a list of endpoints (server/crawl.js): one fetch_runs
++ * row per page, a page's own failure stays the page's, and the source fails only when the site does. While pages are
++ * still due after a run, the next one comes sooner (SOURCES_CRAWL_SOON_MS) than poll_interval_sec.
*/
const { ids } = require('openvibe-contracts');
const { ADAPTERS } = require('./adapters');
const extract = require('./adapters/extract');
++const { createCrawler } = require('./crawl');
const { sha256, decodeBody } = require('./adapters/util');
const FAILURE_STATES = new Set(['http_error', 'timeout', 'robots_denied', 'parse_error', 'rate_limited']);
@@@ -39,6 +44,7 @@ function retryAfterMs(value, now) {
function createIngest({ db, config, registry, items, robots, fetcher, spacer, emitEvent, now = () => Date.now(), log = console, outbox = null }) {
const inflight = new Set();
const { space, setCrawlDelay } = spacer;
++ let stopping = false;
const st = {
insertRun: db.prepare(`INSERT INTO fetch_runs (id, source_key, endpoint_url, trigger, started_at, finished_at, state, http_status,
@@@ -223,6 +229,61 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
return { run, failed: false, success: true, recorded: true };
}
++ const crawler = createCrawler({
++ db, config, items, robots, fetcher, spacer, now, log,
++ record: { baseRun, insertRun: (r) => st.insertRun.run(r), requested: (t, key) => st.requested.run(t, key) },
++ });
++
++ /** Every endpoint in order (feeds, sitemaps, JSON-LD pages, APIs). */
++ async function fetchEndpoints(source, row, trigger, t) {
++ const runs = [];
++ let anyFailure = false;
++ let anySuccess = false;
++ let allAnswered = true; // every endpoint ok or 304: what retirement needs to trust the feed
++ let notBefore = 0;
++ let failures = row.consecutive_failures;
++ for (const endpoint of source.endpoints) {
++ let out;
++ try {
++ out = await fetchEndpoint(source, endpoint, trigger);
++ } catch (err) {
++ // Anything unexpected is still a recorded failure, never a silent gap.
++ const r = baseRun(source, endpoint.url, trigger, t);
++ Object.assign(r, { state: 'http_error', error_code: 'internal', detail: String(err.message).slice(0, 300), finished_at: now() });
++ log.error(`[ingest] ${source.key}: ${err.stack || err}`);
++ out = { run: r, failed: true };
++ }
++ if (out.failed) { anyFailure = true; failures = row.consecutive_failures + 1; }
++ if (out.success) anySuccess = true;
++ else allAnswered = false;
++ if (out.notBefore) notBefore = Math.max(notBefore, out.notBefore);
++ if (!out.recorded) await recordNoItems(source, out.run, { countsAsFailure: out.failed, failures });
++ runs.push(out.run);
++ if (out.run.state === 'rate_limited') break; // the site asked us to stop
++ }
++
++ if (source.retire_after_sec && anySuccess && allAnswered && !anyFailure) {
++ try {
++ const retired = await items.retireUnseen(source, now() - source.retire_after_sec * 1000);
++ if (retired) log.log(`[ingest] ${source.key}: retired ${retired} item(s) the feed stopped showing`);
++ } catch (err) {
++ log.error(`[ingest] ${source.key}: retiring unseen items failed: ${err.stack || err}`);
++ }
++ }
++ const lastState = runs.map(r => r.state).find(s => FAILURE_STATES.has(s)) || runs[runs.length - 1].state;
++ return { runs, anyFailure, anySuccess, notBefore, lastState, more: false };
++ }
++
++ /** The site's frontier, up to pages_per_run pages; one sources.fetch.failed when the site fails, never one per page. */
++ async function crawl(source, row, trigger) {
++ const out = await crawler.run(source, { trigger, stopping: () => stopping });
++ if (out.failedRun) {
++ const failed = out.failedRun;
++ await db.tx(async () => { await failedEvent(source, failed, row.consecutive_failures + 1); });
++ }
++ return { ...out, lastState: out.failedRun ? out.failedRun.state : 'ok' };
++ }
++
// ── One source ───────────────────────────────────────────
/**
@@@ -254,46 +315,17 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
return { runs, rateLimitedUntil: until };
}
-- let anyFailure = false;
-- let anySuccess = false;
-- let allAnswered = true; // every endpoint ok or 304: what retirement needs to trust the feed
-- let notBefore = 0;
-- let failures = row.consecutive_failures;
-- for (const endpoint of source.endpoints) {
-- let out;
-- try {
-- out = await fetchEndpoint(source, endpoint, trigger);
-- } catch (err) {
-- // Anything unexpected is still a recorded failure, never a silent gap.
-- const r = baseRun(source, endpoint.url, trigger, t);
-- Object.assign(r, { state: 'http_error', error_code: 'internal', detail: String(err.message).slice(0, 300), finished_at: now() });
-- log.error(`[ingest] ${key}: ${err.stack || err}`);
-- out = { run: r, failed: true };
-- }
-- if (out.failed) { anyFailure = true; failures = row.consecutive_failures + 1; }
-- if (out.success) anySuccess = true;
-- else allAnswered = false;
-- if (out.notBefore) notBefore = Math.max(notBefore, out.notBefore);
-- if (!out.recorded) await recordNoItems(source, out.run, { countsAsFailure: out.failed, failures });
-- runs.push(out.run);
-- if (out.run.state === 'rate_limited') break; // the site asked us to stop
-- }
--
-- if (source.retire_after_sec && anySuccess && allAnswered && !anyFailure) {
-- try {
-- const retired = await items.retireUnseen(source, now() - source.retire_after_sec * 1000);
-- if (retired) log.log(`[ingest] ${key}: retired ${retired} item(s) the feed stopped showing`);
-- } catch (err) {
-- log.error(`[ingest] ${key}: retiring unseen items failed: ${err.stack || err}`);
-- }
-- }
++ const { runs: done, anyFailure, anySuccess, notBefore, lastState, more } = source.type === 'crawl'
++ ? await crawl(source, row, trigger)
++ : await fetchEndpoints(source, row, trigger, t);
++ runs.push(...done);
const finishedAt = now();
const newFailures = anyFailure ? row.consecutive_failures + 1 : 0;
++ const poll = source.poll_interval_sec * 1000;
const backoff = anyFailure
-- ? Math.min(source.poll_interval_sec * 1000 * 2 ** Math.min(newFailures - 1, 10), Math.max(config.worker.maxBackoffMs, source.poll_interval_sec * 1000))
-- : source.poll_interval_sec * 1000;
-- const lastState = runs.map(r => r.state).find(s => FAILURE_STATES.has(s)) || runs[runs.length - 1].state;
++ ? Math.min(poll * 2 ** Math.min(newFailures - 1, 10), Math.max(config.worker.maxBackoffMs, poll))
++ : more ? Math.min(poll, config.crawl.soonMs) : poll;
await st.finish.run({
key, at: finishedAt, state: lastState, failures: newFailures,
success_at: anySuccess ? finishedAt : null, next_due: finishedAt + backoff, not_before: notBefore,
@@@ -305,7 +337,18 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
}
}
-- return { run, inflight: () => [...inflight] };
++ /** Status of a crawl source's frontier (GET /api/v1/sources/:key/crawl). */
++ async function crawlStatus(key) {
++ const row = await registry.get(key);
++ if (!row) return null;
++ const source = registry.fromRow(row);
++ return source.type === 'crawl' ? await crawler.status(source) : { not_crawl: true };
++ }
++
++ /** Shutting down: a crawl ends after the page in flight instead of working through its budget. */
++ function stop() { stopping = true; }
++
++ return { run, crawlStatus, stop, inflight: () => [...inflight] };
}
module.exports = { createIngest, FAILURE_STATES, retryAfterMs };
ddiff --git a/server/items.js b/server/items.js
iindex 758c056..7c412a8 100644
---- a/server/items.js
++++ b/server/items.js
@@@ -49,6 +49,7 @@ function createItems({ db, emitEvent, now = () => Date.now() }) {
unseen: db.prepare(`SELECT * FROM items WHERE source_key = ? AND removed_at IS NULL AND retrieved_at < ?
ORDER BY retrieved_at, id LIMIT ?`),
seenAgain: db.prepare('UPDATE items SET retrieved_at = ? WHERE source_key = ? AND last_fetch_run_id = ? AND removed_at IS NULL'),
++ ofRun: db.prepare('SELECT * FROM items WHERE source_key = ? AND last_fetch_run_id = ? AND removed_at IS NULL ORDER BY id LIMIT 500'),
};
function view(row) {
@@@ -231,6 +232,20 @@ function createItems({ db, emitEvent, now = () => Date.now() }) {
});
}
++ /**
++ * A crawled page that is gone (404/410 twice) or turned noindex takes its items with it: the live items its last good
++ * read (`runId`) saw are removed with `reason`, sticky like any removal (server/crawl.js). → the number retired
++ */
++ async function retireRun(source, runId, reason) {
++ if (!runId) return 0;
++ return await db.tx(async () => {
++ const rows = await st.ofRun.all(source.key, runId);
++ const t = now();
++ for (const cur of rows) await markRemoved(source, cur, reason, reason, t);
++ return rows.length;
++ });
++ }
++
/**
* A 304 says the endpoint still lists what its last good parse (`runId`) saw: those items count as seen at `at`, so
* retirement does not take them for gone. Only for a source that retires; the others keep the parse's time.
@@@ -266,7 +281,7 @@ function createItems({ db, emitEvent, now = () => Date.now() }) {
return { items: page.map(view), next_after: page.length ? page[page.length - 1].change_seq : after, more };
}
-- return { ingest, remove, retireUnseen, seenAgain, get, list, view, indexDocument, contentHash };
++ return { ingest, remove, retireUnseen, retireRun, seenAgain, get, list, view, indexDocument, contentHash };
}
module.exports = { createItems, contentHash, STAFF_GROUPS };
ddiff --git a/server/registry.js b/server/registry.js
iindex 848dd62..aa24546 100644
---- a/server/registry.js
++++ b/server/registry.js
@@@ -7,12 +7,13 @@
* SOURCES_CRED_ so a registry entry can never point at this service's own secrets), robots and
* terms notes, the rate limit, whether it is enabled, review/sensitivity flags and the default
* indexability downstream products start from. A source cannot be enabled without a terms note:
-- * an adapter existing is not permission to ingest (roadmap anti-goal 26).
++ * an adapter existing is not permission to ingest (roadmap anti-goal 26). A crawl source also
++ * carries how it walks its site (`crawl`, server/crawl.js); its endpoints are the seeds.
*/
const { checkApiMapping } = require('./adapters');
const extract = require('./adapters/extract');
--const TYPES = ['rss', 'atom', 'sitemap', 'jsonld', 'api', 'manual'];
++const TYPES = ['rss', 'atom', 'sitemap', 'jsonld', 'api', 'crawl', 'manual'];
// The categories are sources.source@1's (openvibe-contracts), read from the schema so the two never drift: housing, food
// and jobs since contracts 0.210.0, for Rent and Homes, Food and Work (plan 2026-10-22, track F1).
const CATEGORIES = Object.freeze([...require('openvibe-contracts/contracts/sources/source.v1.json').properties.category.enum]);
@@@ -52,12 +53,41 @@ const bool = (v, d, what) => {
return v;
};
++const CRAWL_FIELDS = ['sitemaps', 'follow_links', 'include', 'exclude', 'max_pages', 'pages_per_run', 'recrawl_sec', 'max_depth', 'require_structured'];
++
++/** A crawl source's walk (sources.source@1 `crawl`), every default filled in so the crawler reads one shape. */
++function validateCrawl(c) {
++ const v = c == null ? {} : c;
++ if (typeof v !== 'object' || Array.isArray(v)) throw new RegistryError('crawl must be an object');
++ for (const k of Object.keys(v)) if (!CRAWL_FIELDS.includes(k)) throw new RegistryError(`unknown crawl field ${k}`);
++ const patterns = (list, what) => {
++ if (list == null) return [];
++ if (!Array.isArray(list) || list.length > 20) throw new RegistryError(`${what} must be an array of at most 20 patterns`);
++ return list.map((p, i) => {
++ if (typeof p !== 'string' || !p || p.length > 200) throw new RegistryError(`${what}[${i}] must be a pattern of 1 to 200 characters`);
++ try { new RegExp(p); } catch (err) { throw new RegistryError(`${what}[${i}] is not a regular expression: ${err.message}`); }
++ return p;
++ });
++ };
++ return {
++ sitemaps: bool(v.sitemaps, true, 'crawl.sitemaps'),
++ follow_links: bool(v.follow_links, false, 'crawl.follow_links'),
++ include: patterns(v.include, 'crawl.include'),
++ exclude: patterns(v.exclude, 'crawl.exclude'),
++ max_pages: intIn(v.max_pages, 1, 200000, 5000, 'crawl.max_pages'),
++ pages_per_run: intIn(v.pages_per_run, 1, 1000, 100, 'crawl.pages_per_run'),
++ recrawl_sec: intIn(v.recrawl_sec, 3600, 30 * 86400, 86400, 'crawl.recrawl_sec'),
++ max_depth: intIn(v.max_depth, 0, 10, 2, 'crawl.max_depth'),
++ require_structured: bool(v.require_structured, true, 'crawl.require_structured'),
++ };
++}
++
/** Validate a full source definition → normalized record (no runtime state). */
function validateSource(input, { maxItemsCap = 12000 } = {}) {
if (!input || typeof input !== 'object' || Array.isArray(input)) throw new RegistryError('body must be a source object');
const known = new Set(['key', 'name', 'type', 'category', 'homepage_url', 'endpoints', 'auth', 'robots_note', 'terms_note', 'license_note',
'min_interval_ms', 'poll_interval_sec', 'stale_after_sec', 'max_items', 'enabled', 'review_required', 'sensitivity',
-- 'default_indexability', 'search_visibility', 'retire_after_sec']);
++ 'default_indexability', 'search_visibility', 'retire_after_sec', 'crawl']);
for (const k of Object.keys(input)) if (!known.has(k)) throw new RegistryError(`unknown field ${k}`);
if (!KEY_RE.test(String(input.key || ''))) throw new RegistryError('key must match ^[a-z0-9][a-z0-9-]{1,63}$');
const name = text(input.name, 200, 'name');
@@@ -139,7 +169,12 @@ function validateSource(input, { maxItemsCap = 12000 } = {}) {
search_visibility: input.search_visibility == null ? null : input.search_visibility,
// Retire what the feed stopped showing (jobs, listings): an hour to a year, or null to keep every item (server/ingest.js).
retire_after_sec: intIn(input.retire_after_sec, 3600, 365 * 86400, null, 'retire_after_sec'),
++ crawl: input.type === 'crawl' ? validateCrawl(input.crawl) : null,
};
++ if (input.type !== 'crawl' && input.crawl != null) throw new RegistryError('crawl is only for crawl sources');
++ // A crawl reaches a page again only when its turn comes, so "not seen for a while" says nothing about a product: a crawl
++ // retires a page's items when the page is gone (404 or 410 twice, server/crawl.js) or asks not to be indexed.
++ if (input.type === 'crawl' && record.retire_after_sec != null) throw new RegistryError('a crawl source retires a page\'s items when the page is gone; retire_after_sec is for feeds');
if (!SENSITIVITY.includes(record.sensitivity)) throw new RegistryError(`sensitivity must be one of ${SENSITIVITY.join('|')}`);
if (!['index', 'noindex'].includes(record.default_indexability)) throw new RegistryError('default_indexability must be index or noindex');
if (record.sensitivity !== 'none' && record.default_indexability === 'index' && record.review_required === false) {
@@@ -157,17 +192,19 @@ function createRegistry({ db, now = () => Date.now(), maxItemsCap = 12000 }) {
all: db.prepare('SELECT * FROM sources ORDER BY key'),
insert: db.prepare(`INSERT INTO sources (key, name, type, category, homepage_url, endpoints, auth, robots_note, terms_note, license_note,
min_interval_ms, poll_interval_sec, stale_after_sec, max_items, enabled, review_required, sensitivity, default_indexability,
-- search_visibility, retire_after_sec, created_at, updated_at, updated_by, next_due_at)
++ search_visibility, retire_after_sec, crawl, created_at, updated_at, updated_by, next_due_at)
VALUES (@key, @name, @type, @category, @homepage_url, @endpoints, @auth, @robots_note, @terms_note, @license_note,
@min_interval_ms, @poll_interval_sec, @stale_after_sec, @max_items, @enabled, @review_required, @sensitivity, @default_indexability,
-- @search_visibility, @retire_after_sec, @created_at, @updated_at, @updated_by, 0)`),
++ @search_visibility, @retire_after_sec, @crawl, @created_at, @updated_at, @updated_by, 0)`),
update: db.prepare(`UPDATE sources SET name = @name, type = @type, category = @category, homepage_url = @homepage_url, endpoints = @endpoints,
auth = @auth, robots_note = @robots_note, terms_note = @terms_note, license_note = @license_note, min_interval_ms = @min_interval_ms,
poll_interval_sec = @poll_interval_sec, stale_after_sec = @stale_after_sec, max_items = @max_items, enabled = @enabled,
review_required = @review_required, sensitivity = @sensitivity, default_indexability = @default_indexability,
-- search_visibility = @search_visibility, retire_after_sec = @retire_after_sec, updated_at = @updated_at, updated_by = @updated_by WHERE key = @key`),
++ search_visibility = @search_visibility, retire_after_sec = @retire_after_sec, crawl = @crawl, updated_at = @updated_at, updated_by = @updated_by
++ WHERE key = @key`),
del: db.prepare('DELETE FROM sources WHERE key = ?'),
itemCount: db.prepare('SELECT COUNT(*) AS n FROM items WHERE source_key = ?'),
++ dropFrontier: db.prepare('DELETE FROM crawl_frontier WHERE source_key = ?'),
resetValidators: db.prepare('DELETE FROM endpoint_state WHERE source_key = ?'),
};
@@@ -176,6 +213,7 @@ function createRegistry({ db, now = () => Date.now(), maxItemsCap = 12000 }) {
...rec,
endpoints: JSON.stringify(rec.endpoints),
auth: JSON.stringify(rec.auth),
++ crawl: rec.crawl ? JSON.stringify(rec.crawl) : null,
enabled: rec.enabled ? 1 : 0,
review_required: rec.review_required ? 1 : 0,
created_at: now(),
@@@ -194,6 +232,7 @@ function createRegistry({ db, now = () => Date.now(), maxItemsCap = 12000 }) {
max_items: row.max_items, enabled: Boolean(row.enabled), review_required: Boolean(row.review_required),
sensitivity: row.sensitivity, default_indexability: row.default_indexability, search_visibility: row.search_visibility,
retire_after_sec: row.retire_after_sec == null ? null : Number(row.retire_after_sec),
++ crawl: row.crawl ? JSON.parse(row.crawl) : null,
};
}
@@@ -256,10 +295,11 @@ function createRegistry({ db, now = () => Date.now(), maxItemsCap = 12000 }) {
if (!row) return null;
if ((await st.itemCount.get(key)).n) throw new RegistryError('the source has items; disable it instead (provenance must stay resolvable)', 'sources.has_items');
await st.del.run(key);
++ await st.dropFrontier.run(key);
return true;
}
return { get: async (k) => await st.get.get(k), all: async () => await st.all.all(), create, patch, remove, view, fromRow, health };
}
--module.exports = { createRegistry, validateSource, RegistryError, TYPES, CATEGORIES };
++module.exports = { createRegistry, validateSource, validateCrawl, RegistryError, TYPES, CATEGORIES };
ddiff --git a/server/robots.js b/server/robots.js
iindex ba4bde4..91f4215 100644
---- a/server/robots.js
++++ b/server/robots.js
@@@ -10,6 +10,8 @@
* complete disallow ("unreachable") — never guessed as allowed
* - cached per origin for SOURCES_ROBOTS_TTL_MS (24 h); Crawl-delay is honoured as a minimum
* gap between requests (non-standard, but a clear wish of the site)
++ * - Sitemap lines (RFC 9309 §2.3.5, outside any group) are kept with the rules: a crawl source
++ * starts from the sitemaps the site names (server/crawl.js)
* A source-adapter name is never permission to fetch (roadmap anti-goal 26): if robots says no,
* the fetch run is `robots_denied` and nothing is requested.
*/
@@@ -17,6 +19,7 @@
function parseRobots(text, agent) {
const token = String(agent).toLowerCase();
const groups = [];
++ const sitemaps = [];
let current = null;
let lastWasAgent = false;
for (const rawLine of String(text).split(/\r\n|\r|\n/)) {
@@@ -36,6 +39,11 @@ function parseRobots(text, agent) {
continue;
}
lastWasAgent = false;
++ if (key === 'sitemap') {
++ // Absolute http(s) URLs only, at most 50: a robots.txt cannot hand the crawler an unbounded list.
++ if (/^https?:\/\//i.test(value) && value.length <= 2048 && sitemaps.length < 50 && !sitemaps.includes(value)) sitemaps.push(value);
++ continue;
++ }
if (!current) continue;
// Bounded: a hostile robots.txt cannot make matching expensive.
if ((key === 'allow' || key === 'disallow') && value.length <= 512 && current.rules.length < 1000) {
@@@ -54,6 +62,7 @@ function parseRobots(text, agent) {
// An empty Disallow (or Allow) path matches nothing: "allow everything".
rules: chosen.flatMap(g => g.rules).filter(r => r.path !== ''),
crawlDelaySec: delays.length ? Math.max(...delays) : null,
++ sitemaps,
};
}
@@@ -70,29 +79,38 @@ function normalizePath(p) {
try { return encodeURI(decodeURI(p)); } catch { return p; }
}
++/** The parsed rules as one function, each rule's pattern compiled once: (pathWithQuery) → allowed. */
++function compileRules(rules) {
++ const compiled = rules.map(r => ({ re: ruleRegex(normalizePath(r.path)), len: r.path.length, allow: r.allow }));
++ return (pathWithQuery) => {
++ const target = normalizePath(pathWithQuery || '/');
++ let best = null;
++ for (const r of compiled) {
++ if (!r.re.test(target)) continue;
++ if (!best || r.len > best.len || (r.len === best.len && r.allow && !best.allow)) best = r;
++ }
++ return best ? best.allow : true;
++ };
++}
++
/** Is `pathWithQuery` allowed by the parsed rules? */
function isAllowed(parsed, pathWithQuery) {
-- const target = normalizePath(pathWithQuery || '/');
-- let best = null;
-- for (const r of parsed.rules) {
-- const re = ruleRegex(normalizePath(r.path));
-- if (!re.test(target)) continue;
-- const len = r.path.length;
-- if (!best || len > best.len || (len === best.len && r.allow && !best.allow)) best = { len, allow: r.allow };
-- }
-- return best ? best.allow : true;
++ return compileRules(parsed.rules)(pathWithQuery);
}
/**
* createRobots({ db, fetcher, agent, ttlMs, maxBytes, now, beforeRequest(host) })
* check(url) → { allowed, reason, crawlDelaySec, source: 'cache'|'fetched' }
++ * matcher(url) → (pathWithQuery) → allowed, for the url's origin (a crawl filtering a sitemap's thousands of URLs)
++ * sitemaps(url) → the Sitemap URLs the origin's robots.txt names
*/
function createRobots({ db, fetcher, agent, ttlMs, maxBytes, now = () => Date.now(), beforeRequest = async () => {} }) {
const get = db.prepare('SELECT * FROM robots_cache WHERE origin = ?');
-- const put = db.prepare(`INSERT INTO robots_cache (origin, fetched_at, expires_at, status, outcome, rules, crawl_delay_sec, detail)
-- VALUES (@origin, @fetched_at, @expires_at, @status, @outcome, @rules, @crawl_delay_sec, @detail)
++ const put = db.prepare(`INSERT INTO robots_cache (origin, fetched_at, expires_at, status, outcome, rules, crawl_delay_sec, detail, sitemaps)
++ VALUES (@origin, @fetched_at, @expires_at, @status, @outcome, @rules, @crawl_delay_sec, @detail, @sitemaps)
ON CONFLICT(origin) DO UPDATE SET fetched_at = excluded.fetched_at, expires_at = excluded.expires_at, status = excluded.status,
-- outcome = excluded.outcome, rules = excluded.rules, crawl_delay_sec = excluded.crawl_delay_sec, detail = excluded.detail`);
++ outcome = excluded.outcome, rules = excluded.rules, crawl_delay_sec = excluded.crawl_delay_sec, detail = excluded.detail,
++ sitemaps = excluded.sitemaps`);
const inflight = new Map();
const warned = new Set();
@@@ -104,15 +122,15 @@ function createRobots({ db, fetcher, agent, ttlMs, maxBytes, now = () => Date.no
const r = await fetcher.fetchUrl(`${origin}/robots.txt`, { headers: { Accept: 'text/plain' }, capBytes: maxBytes });
if (r.status >= 200 && r.status < 300) {
const parsed = parseRobots(r.body.toString('utf8'), agent);
-- row = { status: r.status, outcome: 'parsed', rules: JSON.stringify(parsed.rules), crawl_delay_sec: parsed.crawlDelaySec, detail: null };
++ row = { status: r.status, outcome: 'parsed', rules: JSON.stringify(parsed.rules), crawl_delay_sec: parsed.crawlDelaySec, detail: null, sitemaps: JSON.stringify(parsed.sitemaps) };
} else if (r.status >= 400 && r.status < 500 && r.status !== 429) {
-- row = { status: r.status, outcome: 'unavailable', rules: '[]', crawl_delay_sec: null, detail: `HTTP ${r.status}: no restrictions` };
++ row = { status: r.status, outcome: 'unavailable', rules: '[]', crawl_delay_sec: null, detail: `HTTP ${r.status}: no restrictions`, sitemaps: '[]' };
} else {
-- row = { status: r.status, outcome: 'unreachable', rules: '[]', crawl_delay_sec: null, detail: `HTTP ${r.status}: complete disallow` };
++ row = { status: r.status, outcome: 'unreachable', rules: '[]', crawl_delay_sec: null, detail: `HTTP ${r.status}: complete disallow`, sitemaps: '[]' };
}
} catch (err) {
// Network error, timeout, refused address or an oversized file: refuse, never guess.
-- row = { status: null, outcome: 'unreachable', rules: '[]', crawl_delay_sec: null, detail: `${err.code || 'error'}: ${err.message}`.slice(0, 300) };
++ row = { status: null, outcome: 'unreachable', rules: '[]', crawl_delay_sec: null, detail: `${err.code || 'error'}: ${err.message}`.slice(0, 300), sitemaps: '[]' };
}
// An unreachable robots.txt is retried sooner than a parsed one.
const ttl = row.outcome === 'unreachable' ? Math.min(ttlMs, 15 * 60 * 1000) : ttlMs;
@@@ -128,6 +146,35 @@ function createRobots({ db, fetcher, agent, ttlMs, maxBytes, now = () => Date.no
return { row: await inflight.get(origin), source: 'fetched' };
}
++ function storedRules(row) {
++ try {
++ const rules = JSON.parse(row.rules);
++ return Array.isArray(rules) ? rules : [];
++ } catch (err) {
++ // A corrupt stored row must not throw: treat it as no rules and warn once per origin.
++ if (!warned.has(row.origin)) {
++ warned.add(row.origin);
++ console.warn(`[robots] unreadable stored rules for ${row.origin}: ${err.message}`);
++ }
++ return [];
++ }
++ }
++
++ async function matcher(input) {
++ const { row } = await entry(new URL(input).origin);
++ if (row.outcome === 'unreachable') return () => false;
++ if (row.outcome === 'unavailable') return () => true;
++ return compileRules(storedRules(row));
++ }
++
++ async function sitemaps(input) {
++ const { row } = await entry(new URL(input).origin);
++ try {
++ const list = JSON.parse(row.sitemaps || '[]');
++ return Array.isArray(list) ? list.filter(u => typeof u === 'string') : [];
++ } catch { return []; }
++ }
++
async function check(input) {
const url = new URL(input);
const { row, source } = await entry(url.origin);
@@@ -137,23 +184,11 @@ function createRobots({ db, fetcher, agent, ttlMs, maxBytes, now = () => Date.no
return { allowed: false, reason: `robots.txt unreachable (${row.detail})`, crawlDelaySec: null, source, refusedCode: ['address_refused', 'port_refused', 'bad_url'].includes(code) ? code : null };
}
if (row.outcome === 'unavailable') return { allowed: true, reason: null, crawlDelaySec: null, source };
-- let rules;
-- try {
-- rules = JSON.parse(row.rules);
-- } catch (err) {
-- // A corrupt stored row must not throw: treat it as no rules and warn once per origin.
-- if (!warned.has(row.origin)) {
-- warned.add(row.origin);
-- console.warn(`[robots] unreadable stored rules for ${row.origin}: ${err.message}`);
-- }
-- rules = [];
-- }
-- const parsed = { rules: Array.isArray(rules) ? rules : [] };
-- const allowed = isAllowed(parsed, url.pathname + url.search);
++ const allowed = isAllowed({ rules: storedRules(row) }, url.pathname + url.search);
return { allowed, reason: allowed ? null : `disallowed by ${url.origin}/robots.txt`, crawlDelaySec: row.crawl_delay_sec, source };
}
-- return { check, forget: async (origin) => await db.prepare('DELETE FROM robots_cache WHERE origin = ?').run(origin) };
++ return { check, matcher, sitemaps, forget: async (origin) => await db.prepare('DELETE FROM robots_cache WHERE origin = ?').run(origin) };
}
--module.exports = { parseRobots, isAllowed, createRobots };
++module.exports = { parseRobots, isAllowed, compileRules, createRobots };
ddiff --git a/server/scheduler.js b/server/scheduler.js
iindex 4d08c49..368e84c 100644
---- a/server/scheduler.js
++++ b/server/scheduler.js
@@@ -2,20 +2,31 @@
/**
* In-process scheduler: every SOURCES_TICK_MS, start runs for enabled, fetched sources whose
* next_due_at and not_before have passed, oldest first, up to SOURCES_MAX_CONCURRENT runs at a
-- * time. ingest.run() itself refuses a second run of a source already in flight.
++ * time, of which at most SOURCES_MAX_CONCURRENT_CRAWLS crawls (a crawl run lasts minutes, a feed's
++ * seconds). ingest.run() itself refuses a second run of a source already in flight.
*/
function createScheduler({ db, ingest, config, now = () => Date.now(), log = console }) {
-- const due = db.prepare(`SELECT key FROM sources WHERE enabled = 1 AND type != 'manual'
++ const due = db.prepare(`SELECT key, type FROM sources WHERE enabled = 1 AND type != 'manual'
AND next_due_at <= @now AND not_before <= @now ORDER BY next_due_at, key LIMIT @limit`);
let timer = null;
-- const running = new Set();
++ const running = new Map(); // key → type
async function tick() {
const free = config.worker.maxConcurrent - running.size;
if (free <= 0) return [];
-- const keys = (await due.all({ now: now(), limit: free + running.size })).map(r => r.key).filter(k => !running.has(k)).slice(0, free);
++ let crawls = [...running.values()].filter(t => t === 'crawl').length;
++ const keys = [];
++ for (const r of await due.all({ now: now(), limit: free + running.size + config.crawl.maxConcurrent })) {
++ if (keys.length >= free) break;
++ if (running.has(r.key)) continue;
++ if (r.type === 'crawl') {
++ if (crawls >= config.crawl.maxConcurrent) continue;
++ crawls++;
++ }
++ keys.push(r.key);
++ running.set(r.key, r.type);
++ }
for (const key of keys) {
-- running.add(key);
ingest.run(key, { trigger: 'schedule' })
.catch(err => log.error(`[scheduler] ${key}: ${err.stack || err}`))
.finally(() => running.delete(key));
@@@ -32,12 +43,13 @@ function createScheduler({ db, ingest, config, now = () => Date.now(), log = con
async function stop() {
if (timer) clearInterval(timer);
timer = null;
++ if (ingest.stop) ingest.stop();
// let runs in flight finish (each is bounded by the fetch timeout)
const deadline = Date.now() + 15000;
while (running.size && Date.now() < deadline) await new Promise(r => setTimeout(r, 50));
}
-- return { start, stop, tick, running: () => Boolean(timer), active: () => [...running] };
++ return { start, stop, tick, running: () => Boolean(timer), active: () => [...running.keys()] };
}
module.exports = { createScheduler };
ddiff --git a/test/crawl.test.js b/test/crawl.test.js
nnew file mode 100644
iindex 0000000..fea6cca
---- /dev/null
++++ b/test/crawl.test.js
@@@ -0,0 +1,333 @@
++'use strict';
++/**
++ * Crawl sources against stub sites: the registry's crawl settings, sitemaps from robots.txt (indexes, gzip, lastmod),
++ * include/exclude and robots.txt on every page, JSON-LD products kept with provenance, recrawl and its growing wait,
++ * noindex, canonical duplicates, 404 twice retires a page's items, links followed to max_depth (nofollow and other
++ * hosts skipped), page items when structured data is not required, the site failing as the source's failure (robots
++ * unreachable, 429), the run budget, the scheduler's crawl limit, and the API.
++ */
++const assert = require('assert');
++const zlib = require('zlib');
++const { boot, site, sourceDef, request, serviceToken, suite } = require('./helpers');
++const { validateSource } = require('../server/registry');
++const page = require('../server/adapters/page');
++
++const t = suite('crawl');
++const ADMIN = serviceToken('network', ['sources.source.manage', 'sources.source.read']);
++const clock = { offset: 0 };
++const DAY = 86400 * 1000;
++let svc;
++let web;
++let price;
++let gonePage;
++
++const product = (name, sku, p) => `<!doctype html><html lang="en"><head><title>${name}</title>
++<script type="application/ld+json">${JSON.stringify({ '@context': 'https://schema.org', '@type': 'Product', name, sku,
++ offers: { '@type': 'Offer', price: p, priceCurrency: 'USD', availability: 'https://schema.org/InStock' } })}</script></head><body><h1>${name}</h1></body></html>`;
++const html = (head, body = '') => ({ headers: { 'Content-Type': 'text/html; charset=utf-8' }, body: `<!doctype html><html><head>${head}</head><body>${body}</body></html>` });
++const xml = (body) => ({ headers: { 'Content-Type': 'application/xml' }, body });
++const urlset = (urls) => `<?xml version="1.0"?><urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">${urls.map(u =>
++ `<url><loc>${u.loc}</loc>${u.lastmod ? `<lastmod>${u.lastmod}</lastmod>` : ''}</url>`).join('')}</urlset>`;
++
++const frontier = async (key) => await svc.db.prepare('SELECT * FROM crawl_frontier WHERE source_key = ? ORDER BY url').all(key);
++const rowOf = async (key, path) => (await frontier(key)).find(r => r.url === `${web.origin}${path}`);
++const itemsOf = async (key) => await svc.db.prepare('SELECT * FROM items WHERE source_key = ? ORDER BY identity').all(key);
++const runsOf = async (key) => await svc.db.prepare('SELECT * FROM fetch_runs WHERE source_key = ? ORDER BY rid').all(key);
++const events = async (type) => (await svc.db.prepare('SELECT envelope FROM service_outbox ORDER BY id').all()).map(r => r.envelope).filter(e => e.event_type === type);
++const run = async (key) => await svc.ingest.run(key, { trigger: 'manual' });
++const crawlDef = (o = {}) => sourceDef({ type: 'crawl', category: 'deals', endpoints: [`${web.origin}/`], min_interval_ms: 0, ...o });
++
++t('boot', async () => {
++ svc = await boot({ now: () => Date.now() + clock.offset });
++ price = '19.99';
++ gonePage = false;
++ web = await site({
++ '/robots.txt': () => ({ body: `User-agent: *\nDisallow: /private/\nSitemap: ${web.origin}/sitemap-index.xml\n` }),
++ '/sitemap-index.xml': () => xml(`<?xml version="1.0"?><sitemapindex xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
++ <sitemap><loc>${web.origin}/sitemap-products.xml.gz</loc></sitemap><sitemap><loc>https://elsewhere.example/sitemap.xml</loc></sitemap></sitemapindex>`),
++ '/sitemap-products.xml.gz': () => ({
++ headers: { 'Content-Type': 'application/x-gzip' },
++ body: zlib.gzipSync(urlset([
++ { loc: `${web.origin}/products/lamp`, lastmod: '2026-10-01' },
++ { loc: `${web.origin}/products/chair` },
++ { loc: `${web.origin}/products/gone` },
++ { loc: `${web.origin}/products/cart-thing` },
++ { loc: `${web.origin}/private/secret` },
++ { loc: `${web.origin}/about` },
++ { loc: `${web.origin}/products/plain` },
++ { loc: `${web.origin}/products/hidden` },
++ { loc: `${web.origin}/products/lamp-copy` },
++ { loc: 'https://elsewhere.example/products/x' },
++ ])),
++ }),
++ '/': () => html('<title>Shop</title>'),
++ '/products/lamp': (req) => {
++ const body = product('Desk lamp', 'L-1', price);
++ const etag = `"lamp-${price}"`;
++ if (req.headers['if-none-match'] === etag) return { status: 304 };
++ return { headers: { 'Content-Type': 'text/html', ETag: etag }, body };
++ },
++ '/products/chair': () => ({ headers: { 'Content-Type': 'text/html' }, body: product('Chair', 'C-1', '49.00') }),
++ '/products/gone': () => (gonePage ? { status: 404, body: 'gone' } : { headers: { 'Content-Type': 'text/html' }, body: product('Old thing', 'G-1', '5.00') }),
++ '/products/cart-thing': () => ({ headers: { 'Content-Type': 'text/html' }, body: product('Never fetched', 'X', '1.00') }),
++ '/products/plain': () => html('<title>Plain page</title><meta name="description" content="A page without structured data.">'),
++ '/products/hidden': () => ({ headers: { 'Content-Type': 'text/html', 'X-Robots-Tag': 'noindex' }, body: product('Hidden', 'H-1', '2.00') }),
++ '/products/lamp-copy': () => html(`<link rel="canonical" href="${web.origin}/products/lamp"><title>Lamp (copy)</title>`),
++ '/private/secret': () => ({ headers: { 'Content-Type': 'text/html' }, body: product('Secret', 'S', '1.00') }),
++ });
++});
++
++t('the registry fills a crawl source\'s defaults and refuses what does not belong', () => {
++ const rec = validateSource({ ...crawlDef({ key: 'c-defaults' }) });
++ assert.deepStrictEqual(rec.crawl, {
++ sitemaps: true, follow_links: false, include: [], exclude: [], max_pages: 5000, pages_per_run: 100,
++ recrawl_sec: 86400, max_depth: 2, require_structured: true,
++ });
++ assert.strictEqual(validateSource(sourceDef({ key: 'c-rss', endpoints: [`${web.origin}/f.xml`] })).crawl, null);
++ assert.throws(() => validateSource(sourceDef({ key: 'c-rss2', endpoints: [`${web.origin}/f.xml`], crawl: {} })), /only for crawl sources/);
++ assert.throws(() => validateSource(crawlDef({ key: 'c-bad', crawl: { include: ['(['] } })), /not a regular expression/);
++ assert.throws(() => validateSource(crawlDef({ key: 'c-bad2', crawl: { speed: 9 } })), /unknown crawl field speed/);
++ assert.throws(() => validateSource(crawlDef({ key: 'c-bad3', crawl: { recrawl_sec: 60 } })), /recrawl_sec/);
++ assert.throws(() => validateSource(crawlDef({ key: 'c-bad4', retire_after_sec: 86400 })), /retire_after_sec is for feeds/);
++});
++
++t('a crawl reads the sitemaps robots.txt names and keeps the products their pages publish', async () => {
++ await svc.registry.create(crawlDef({ key: 'shop', crawl: { include: ['^/products/', '^/private/', '^/about'], exclude: ['cart'] } }), 'test');
++ const out = await run('shop');
++ const rows = await frontier('shop');
++ const by = Object.fromEntries(rows.map(r => [r.url.replace(web.origin, ''), r]));
++ // robots.txt's sitemap, the gzip child; never the other host's
++ assert.strictEqual(by['/sitemap-index.xml'].found_via, 'robots');
++ assert.strictEqual(by['/sitemap-index.xml'].state, 'read');
++ assert.strictEqual(by['/sitemap-products.xml.gz'].state, 'read');
++ assert.ok(!rows.some(r => r.url.includes('elsewhere.example')), 'nothing from another host');
++ // include/exclude and robots.txt decide what joins the frontier
++ assert.ok(!by['/products/cart-thing'], 'excluded by pattern');
++ assert.ok(!by['/private/secret'], 'disallowed by robots.txt');
++ assert.strictEqual(web.hits('/private/secret').length, 0);
++ assert.strictEqual(web.hits('/products/cart-thing').length, 0);
++ // the seed itself is a page with nothing structured: dropped
++ assert.strictEqual(by['/'].state, 'unstructured');
++ assert.strictEqual(by['/'].next_fetch_at, null);
++
++ assert.strictEqual(by['/products/lamp'].state, 'kept');
++ assert.strictEqual(by['/products/chair'].state, 'kept');
++ assert.strictEqual(by['/products/plain'].state, 'unstructured');
++ assert.strictEqual(by['/products/hidden'].state, 'noindex');
++ assert.strictEqual(by['/products/lamp-copy'].state, 'duplicate');
++ assert.strictEqual(by['/about'].state, 'missing', 'a first 404 is checked again tomorrow');
++ const items = await itemsOf('shop');
++ assert.deepStrictEqual(items.map(i => i.title).sort(), ['Chair', 'Desk lamp', 'Old thing']);
++ const lamp = items.find(i => i.title === 'Desk lamp');
++ assert.strictEqual(lamp.kind, 'product');
++ assert.strictEqual(JSON.parse(lamp.fields).offers[0].price, '19.99');
++ assert.strictEqual(lamp.parser_version, 'jsonld@1');
++ assert.strictEqual(lamp.last_fetch_run_id, by['/products/lamp'].last_ok_run_id);
++ const runs = await runsOf('shop');
++ assert.ok(runs.every(r => r.state !== 'parse_error'));
++ assert.strictEqual(out.runs.length, runs.length, 'one run row per request');
++ assert.strictEqual((await svc.registry.get('shop')).last_state, 'ok');
++ assert.strictEqual((await svc.registry.get('shop')).consecutive_failures, 0);
++});
++
++t('nothing due: a run requests nothing and the crawl counts as current', async () => {
++ const before = web.requests.length;
++ const out = await run('shop');
++ assert.deepStrictEqual(out.runs, []);
++ assert.strictEqual(web.requests.length, before);
++ const row = await svc.registry.get('shop');
++ assert.strictEqual(row.last_state, 'ok');
++});
++
++t('recrawl: a changed price is a new revision, an unchanged page waits longer, a 304 refreshes its items', async () => {
++ clock.offset += 2 * DAY;
++ price = '17.49';
++ await run('shop');
++ const lamp = (await itemsOf('shop')).find(i => i.title === 'Desk lamp');
++ assert.strictEqual(lamp.revision, 2);
++ assert.strictEqual(JSON.parse(lamp.fields).offers[0].price, '17.49');
++ const lampRow = await rowOf('shop', '/products/lamp');
++ assert.strictEqual(Number(lampRow.interval_sec), 86400, 'a change keeps the base wait');
++ const chairRow = await rowOf('shop', '/products/chair');
++ assert.strictEqual(Number(chairRow.interval_sec), 129600, 'the same page waits 1.5 times longer');
++
++ clock.offset += 2 * DAY;
++ const etagHits = web.hits('/products/lamp').length;
++ await run('shop');
++ assert.strictEqual(web.hits('/products/lamp').length, etagHits + 1);
++ assert.strictEqual(web.hits('/products/lamp').slice(-1)[0].headers['if-none-match'], '"lamp-17.49"');
++ const lampRuns = (await runsOf('shop')).filter(r => r.endpoint_url.endsWith('/products/lamp'));
++ assert.strictEqual(lampRuns[lampRuns.length - 1].state, 'not_modified');
++ const lamp2 = (await itemsOf('shop')).find(i => i.title === 'Desk lamp');
++ assert.ok(Number(lamp2.retrieved_at) > Number(lamp.retrieved_at), 'a 304 counts as seeing the item again');
++});
++
++t('a page that answers 404 twice is gone and its items are retired', async () => {
++ gonePage = true;
++ clock.offset += 5 * DAY;
++ await run('shop');
++ let row = await rowOf('shop', '/products/gone');
++ assert.strictEqual(row.state, 'missing');
++ assert.strictEqual((await itemsOf('shop')).find(i => i.title === 'Old thing').removed_at, null, 'one 404 is not enough');
++ clock.offset += 2 * DAY;
++ await run('shop');
++ row = await rowOf('shop', '/products/gone');
++ assert.strictEqual(row.state, 'gone');
++ const old = (await itemsOf('shop')).find(i => i.title === 'Old thing');
++ assert.ok(old.removed_at);
++ assert.strictEqual(old.removed_reason, 'gone_from_source');
++ assert.strictEqual((await svc.registry.get('shop')).consecutive_failures, 0, 'a missing page is not the source failing');
++ assert.strictEqual((await events('sources.fetch.failed')).filter(e => e.payload.source_key === 'shop').length, 0);
++});
++
++t('a newer lastmod in the sitemap makes a known page due again', async () => {
++ const row = await rowOf('shop', '/products/lamp');
++ const r = await svc.db.prepare('UPDATE crawl_frontier SET next_fetch_at = ?, last_fetch_at = ? WHERE source_key = ? AND url = ?')
++ .run(Date.now() + clock.offset + 10 * DAY, Date.parse('2026-09-01'), 'shop', row.url);
++ assert.strictEqual(r.changes, 1);
++ // the sitemap's lastmod (2026-10-01) is after that fetch: the next read of the sitemap makes the page due
++ await svc.db.prepare("UPDATE crawl_frontier SET next_fetch_at = 0, body_hash = NULL WHERE source_key = 'shop' AND kind = 'sitemap'").run();
++ await run('shop');
++ const after = await rowOf('shop', '/products/lamp');
++ assert.ok(Number(after.last_fetch_at) > Date.parse('2026-09-01'), 'fetched again in the same run');
++});
++
++t('links are followed to max_depth on the crawl\'s hosts; nofollow and other hosts are not', async () => {
++ const hub = await site({
++ '/robots.txt': () => ({ body: 'User-agent: *\nAllow: /\n' }),
++ '/': () => html('<title>Hub</title>', `<a href="/c/1">one</a> <a href="/c/2" rel="nofollow">two</a> <a href="https://elsewhere.example/x">out</a> <a href="/c/1#top">again</a>`),
++ '/c/1': () => html('<title>Category</title>', '<a href="/p/1">product</a>'),
++ '/p/1': () => ({ headers: { 'Content-Type': 'text/html' }, body: product('Linked', 'P-1', '3.00') }),
++ '/c/2': () => html('<title>Two</title>'),
++ });
++ await svc.registry.create(sourceDef({ key: 'hub', type: 'crawl', category: 'deals', endpoints: [`${hub.origin}/`], min_interval_ms: 0,
++ crawl: { sitemaps: false, follow_links: true, max_depth: 1 } }), 'test');
++ await run('hub');
++ const rows = Object.fromEntries((await svc.db.prepare('SELECT * FROM crawl_frontier WHERE source_key = ?').all('hub')).map(r => [r.url.replace(hub.origin, ''), r]));
++ assert.strictEqual(rows['/'].state, 'hub');
++ assert.ok(rows['/'].next_fetch_at != null, 'a hub is crawled again');
++ assert.strictEqual(Number(rows['/c/1'].depth), 1);
++ assert.strictEqual(rows['/c/1'].state, 'unstructured', 'at max_depth its links are not followed, so it is no hub');
++ assert.ok(!rows['/c/2'], 'rel=nofollow');
++ assert.ok(!rows['/p/1'], 'beyond max_depth');
++ assert.ok(!Object.keys(rows).some(k => k.includes('elsewhere')));
++ assert.strictEqual(hub.hits('/sitemap.xml').length, 0, 'sitemaps: false');
++ await hub.close();
++});
++
++t('a source that does not require structured data keeps a page item from the page\'s metadata', async () => {
++ const blog = await site({
++ '/robots.txt': () => ({ body: '' }),
++ '/post': () => html('<title>A post</title><meta name="description" content="What the post says it is about."><meta property="og:image" content="/img.png"><meta property="article:published_time" content="2026-10-20T10:00:00Z">'),
++ '/quiet': () => html('<title>Quiet</title><meta name="robots" content="noindex">'),
++ '/other-bot': () => ({ headers: { 'Content-Type': 'text/html', 'X-Robots-Tag': 'otherbot: noindex' }, body: '<html><head><title>Other bot</title></head></html>' }),
++ });
++ await svc.registry.create(sourceDef({ key: 'blog', type: 'crawl', category: 'blog', endpoints: [`${blog.origin}/post`, `${blog.origin}/quiet`, `${blog.origin}/other-bot`], min_interval_ms: 0,
++ crawl: { sitemaps: false, require_structured: false } }), 'test');
++ await run('blog');
++ const items = await itemsOf('blog');
++ assert.deepStrictEqual(items.map(i => i.title).sort(), ['A post', 'Other bot']);
++ const post = items.find(i => i.title === 'A post');
++ assert.strictEqual(post.kind, 'page');
++ assert.strictEqual(post.summary, 'What the post says it is about.');
++ assert.strictEqual(post.published_at, '2026-10-20T10:00:00.000Z');
++ assert.strictEqual(JSON.parse(post.fields).image, `${blog.origin}/img.png`);
++ assert.strictEqual(post.parser_version, 'page@1');
++ assert.ok(!items.some(i => i.title === 'Quiet'), 'noindex');
++ await blog.close();
++});
++
++t('X-Robots-Tag scoped to another crawler does not apply; scoped to ours it does', () => {
++ assert.deepStrictEqual(page.headerDirectives('otherbot: noindex', 'OpenVibeSources'), { noindex: false, nofollow: false });
++ assert.deepStrictEqual(page.headerDirectives('openvibesources: noindex, nofollow', 'OpenVibeSources'), { noindex: true, nofollow: true });
++ assert.deepStrictEqual(page.headerDirectives(['none'], 'OpenVibeSources'), { noindex: true, nofollow: true });
++ assert.deepStrictEqual(page.headerDirectives('unavailable_after: 2030-01-01', 'OpenVibeSources'), { noindex: false, nofollow: false });
++});
++
++t('the site failing is the source failing: robots.txt unreachable, then a 429', async () => {
++ let robotsStatus = 500;
++ let limited = false;
++ const down = await site({
++ '/robots.txt': () => (robotsStatus === 200 ? { body: '' } : { status: robotsStatus, body: 'oops' }),
++ '/a': () => (limited ? { status: 429, headers: { 'Retry-After': '120' }, body: 'slow down' } : { headers: { 'Content-Type': 'text/html' }, body: product('A', 'A', '1.00') }),
++ '/b': () => ({ headers: { 'Content-Type': 'text/html' }, body: product('B', 'B', '1.00') }),
++ });
++ await svc.registry.create(sourceDef({ key: 'down', type: 'crawl', category: 'deals', endpoints: [`${down.origin}/a`, `${down.origin}/b`], min_interval_ms: 0, crawl: { sitemaps: false } }), 'test');
++ let out = await run('down');
++ assert.strictEqual(out.runs.length, 1, 'the first page tells: the run stops');
++ assert.strictEqual(out.runs[0].state, 'robots_denied');
++ assert.strictEqual(down.hits('/a').length, 0);
++ let row = await svc.registry.get('down');
++ assert.strictEqual(row.consecutive_failures, 1);
++ assert.strictEqual(row.last_state, 'robots_denied');
++ assert.strictEqual((await events('sources.fetch.failed')).filter(e => e.payload.source_key === 'down').length, 1, 'one event for the run');
++
++ robotsStatus = 200;
++ limited = true;
++ await svc.robots.forget(down.origin);
++ await svc.db.prepare("UPDATE sources SET last_request_at = NULL, not_before = 0 WHERE key = 'down'").run();
++ out = await run('down');
++ assert.strictEqual(out.runs[out.runs.length - 1].state, 'rate_limited');
++ assert.strictEqual(down.hits('/b').length, 0, 'nothing after the site asked us to wait');
++ row = await svc.registry.get('down');
++ assert.strictEqual(row.consecutive_failures, 2);
++ assert.ok(Number(row.not_before) >= Date.now() + clock.offset + 100 * 1000);
++ await down.close();
++});
++
++t('pages_per_run caps one run, and pages still due bring the next run sooner', async () => {
++ const many = await site({
++ '/robots.txt': () => ({ body: '' }),
++ '*': (req, res, ctx) => ({ headers: { 'Content-Type': 'text/html' }, body: product(`Item ${ctx.url.pathname}`, ctx.url.pathname, '1.00') }),
++ });
++ await svc.registry.create(sourceDef({ key: 'many', type: 'crawl', category: 'deals', poll_interval_sec: 3600, min_interval_ms: 0,
++ endpoints: [1, 2, 3, 4, 5].map(n => `${many.origin}/p/${n}`), crawl: { sitemaps: false, pages_per_run: 2 } }), 'test');
++ const out = await run('many');
++ assert.strictEqual(out.runs.length, 2);
++ const row = await svc.registry.get('many');
++ const soon = Number(row.next_due_at) - Number(row.last_run_at);
++ assert.ok(soon <= 60 * 1000, `next run in ${soon} ms, not the poll interval`);
++ await many.close();
++});
++
++t('the API shows the frontier and starts a crawl without waiting for it', async () => {
++ let r = await request(svc.base, 'GET', '/api/v1/sources/shop/crawl', { token: ADMIN });
++ assert.strictEqual(r.status, 200);
++ assert.ok(r.body.crawl.pages >= 8);
++ assert.strictEqual(r.body.crawl.pages_by_state.kept, 2);
++ assert.strictEqual(r.body.crawl.pages_by_state.gone, 2, '/products/gone and /about');
++ assert.strictEqual(r.body.crawl.sitemaps_by_state.read, 2);
++ assert.ok(r.body.crawl.recent.length > 0);
++ assert.ok(r.body.crawl.recent[0].url.startsWith(web.origin));
++ r = await request(svc.base, 'GET', '/api/v1/sources/nope/crawl', { token: ADMIN });
++ assert.strictEqual(r.status, 404);
++ await svc.registry.create(sourceDef({ key: 'feedish', endpoints: [`${web.origin}/feed.xml`] }), 'test');
++ r = await request(svc.base, 'GET', '/api/v1/sources/feedish/crawl', { token: ADMIN });
++ assert.strictEqual(r.status, 409);
++ r = await request(svc.base, 'POST', '/api/v1/sources/shop/fetch', { token: ADMIN });
++ assert.strictEqual(r.status, 202);
++ assert.strictEqual(r.body.started, true);
++ r = await request(svc.base, 'GET', '/api/v1/sources/shop', { token: ADMIN });
++ assert.strictEqual(r.body.source.crawl.max_pages, 5000);
++ // let the background run finish before the next test
++ for (let i = 0; i < 100 && svc.ingest.inflight().includes('shop'); i++) await new Promise(res => setTimeout(res, 20));
++});
++
++t('the scheduler runs at most SOURCES_MAX_CONCURRENT_CRAWLS crawls at once', async () => {
++ const two = await boot({ env: { SOURCES_MAX_CONCURRENT_CRAWLS: '1', SOURCES_MAX_CONCURRENT: '4' } });
++ const slow = await site({ '/robots.txt': () => ({ body: '' }), '*': () => ({ delayMs: 300, headers: { 'Content-Type': 'text/html' }, body: product('S', 'S', '1.00') }) });
++ for (const k of ['crawl-a', 'crawl-b']) await two.registry.create(sourceDef({ key: k, type: 'crawl', category: 'deals', endpoints: [`${slow.origin}/${k}`], min_interval_ms: 0, crawl: { sitemaps: false } }), 'test');
++ await two.registry.create(sourceDef({ key: 'feed-c', endpoints: [`${slow.origin}/feed.xml`] }), 'test');
++ const started = await two.scheduler.tick();
++ assert.deepStrictEqual(started.sort(), ['crawl-a', 'feed-c']);
++ for (let i = 0; i < 200 && two.scheduler.active().length; i++) await new Promise(res => setTimeout(res, 20));
++ await two.stop();
++ await slow.close();
++});
++
++t('teardown', async () => {
++ await svc.stop();
++ await web.close();
++});
++
++t.run();