openvibers/openvibe.sources Commits

Coverage round 2: bulk datasets, api filter, HN Who is hiring resolver and four disabled seeds (#151)

OP OpenVibers committed 14 days ago · d2503c795e · 22 files, +1309 −80

A bulk source streams one CSV, JSONL or TSV dataset file with a byte cap and a per-run row limit, stopping the download at the limit; api@1 gets filter as an alias of where; and an hn-whoishiring resolver walks the monthly thread's top-level comments only, strips emails and phones, and links the full comment instead of storing it. Adds the four disabled seeds (open-food-facts, bls-cpi-rent, themealdb, hn-whoishiring) with their re-checked robots, terms and licence notes.

Changes

ddiff --git a/README.md b/README.md
iindex 9d5e341..e1aafe2 100644
---- a/README.md
++++ b/README.md
@@@ -25,7 +25,7 @@ decide — under their own review and indexability rules — what, if anything,
```bash
npm install
cp .env.example .env
--npm run seed # the 54 sources in seeds/sources.json, created disabled
++npm run seed # the 65 sources in seeds/sources.json, created disabled
npm run dev # http://127.0.0.1:4720
npm test # every test/*.test.js against local stub sites; no internet
```
@@@ -53,9 +53,9 @@ start and flushes on stop.
## The registry
--A source records: `key`, `name`, `type` (`rss|atom|sitemap|jsonld|api|crawl|manual`), `category`
++A source records: `key`, `name`, `type` (`rss|atom|sitemap|jsonld|api|bulk|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 —
++provider; for `bulk`: one dataset file with a format, a byte cap and a row limit), `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
secrets), `robots_note`, `terms_note`, `license_note`, the rate limit (`min_interval_ms`),
`poll_interval_sec`, `stale_after_sec`, `max_items`, `enabled`, `review_required`,
@@@ -83,7 +83,7 @@ An in-process scheduler starts runs for enabled sources whose poll is due (at mo
CGNAT, metadata and v4-mapped/NAT64/6to4 forms are refused), manual redirects (≤ 5, each
re-checked), conditional GET (`If-None-Match`/`If-Modified-Since`), one deadline for the whole
response (`SOURCES_FETCH_TIMEOUT_MS`), a byte cap on the decoded body (`SOURCES_MAX_BYTES`).
--5. **Parse** with the adapter; on success, in one transaction: the run row, the items (created /
++5. **Parse** with the adapter (a `bulk` dataset is parsed chunk by chunk and the stream is stopped at its row limit; a resolver walks its JSON documents); on success, in one transaction: the run row, the items (created /
new revision if the content hash changed / retrieval time refreshed if unchanged), the
validators, and the events.
@@@ -105,7 +105,8 @@ succeeded. Health status: `healthy | stale | failing | never_fetched | disabled
| `rss`, `atom` (`feed@3`) | RSS 2.0, RSS 1.0/RDF, Atom 1.0 | guid/id identity, else link; text-only summary capped at 1000 chars; a deal feed's own offer elements (DealNews' `dealnews:price` with its currency, `dealnews:retailer`, `dealnews:expires`) become `fields.price`, `currency`, `seller` and `valid_until` exactly as stated (the item stays an article; nothing is read out of the headline); its `dealnews:category` fills `fields.categories` when the item has no `<category>`, and `dealnews:staffPick` is `fields.staff_pick` (true/false, absent when not stated) |
| `sitemap` (`sitemap@1`) | `<urlset>` (+ Google News extension), `<sitemapindex>` | index children are listed, not followed |
| `jsonld` (`jsonld@2`) | 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`; `list_price` from a ListPrice/StrikethroughPrice/MSRP/SRP `priceSpecification`, as stated |
--| `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 |
++| `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`, `where`/`filter`; epoch fields only where mapped as such; `resolver: "hn-whoishiring"` walks the Hacker News thread's JSON documents instead |
++| `bulk` (`bulk@1`) | one dataset file, streamed (CSV, JSONL or TSV) | `max_bytes` is the size cap, `max_rows` the per-run row limit; the stream stops at the limit, so a multi-GB file is never held in memory |
| `crawl` (`jsonld@2`, `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 |
@@@ -151,7 +152,7 @@ A value is literal text or `$rec.<path>`, read from the record itself. No match
"price_monthly": "pricings[mode=default,commitment=0,intervalUnit=month,interval=1].price" }
```
--`max_bytes` (an integer from 1 KiB to 32 MiB) sets one endpoint's body cap instead of `SOURCES_MAX_BYTES`. OVHcloud's
++`max_bytes` (an integer from 1 KiB to 64 MiB) sets one endpoint's body cap instead of `SOURCES_MAX_BYTES`. OVHcloud's
catalog is about 12 MB.
`timeout_ms` (an integer from 1 s to 120 s) sets one endpoint's deadline instead of `SOURCES_FETCH_TIMEOUT_MS`. NYC Open
@@@ -339,7 +340,7 @@ for the N-1 rollback window.
## Seeds
--[seeds/sources.json](seeds/sources.json) holds the 54 registered sources, each with its robots, terms and licence notes (the production rows are the record; the terms notes match them as of the 2026-10-23 review).
++[seeds/sources.json](seeds/sources.json) holds the 65 registered sources, each with its robots, terms and licence notes (the production rows are the record; the terms notes match them as of the 2026-10-23 review).
`npm run seed` creates the ones that do not exist yet, **all disabled**, and never changes an existing one; a person
enables a source after reading its terms (`PATCH /api/v1/sources/:key {"enabled": true}`), and the fetcher re-checks
robots.txt on every run. Production on 2026-10-22:
ddiff --git a/docs/contracts-proposal/contracts/sources/source.v1.json b/docs/contracts-proposal/contracts/sources/source.v1.json
iindex 45e08bf..dec941e 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",
++ "bulk",
"crawl",
"manual"
]
@@@ -84,9 +85,11 @@
"enum": [
"json",
"xml",
-- "tsv"
++ "tsv",
++ "csv",
++ "jsonl"
],
-- "description": "json (default), xml, or tsv: a tab-separated file whose first line names the columns, read as an array of records keyed by those names (items_path is \"\"; BLS series files, Sources 2026-10-24)."
++ "description": "json (default), xml, tsv (a tab-separated file whose first line names the columns; BLS series files, Sources 2026-10-24), or, for a bulk source, csv, jsonl or tsv: one dataset file read as a stream, row by row (Sources, 2026-10-31)."
},
"duplicates": {
"enum": [
@@@ -147,6 +150,16 @@
},
"description": "Records kept only when every path equals its value (or one of an array of values), or, for { pattern }, is text the regular expression matches; others are left out before max_items counts. Paths follow the field rules: relative to the record, $key, an absolute $root.<path> into the response (a {$key} segment there is the record's map key), * for an object's single child, and key[field=value,...] selectors (Sources#39, Sources#51)."
},
++ "filter": {
++ "$ref": "#/properties/endpoints/items/properties/where",
++ "description": "The same row conditions as where; the name the BLS CPI rent seed uses (Sources, 2026-10-31). When both are given, both apply."
++ },
++ "resolver": {
++ "enum": [
++ "hn-whoishiring"
++ ],
++ "description": "An api endpoint whose records are walked from several JSON URLs instead of one items_path. hn-whoishiring reads user/whoishiring.json, the latest \"Who is hiring\" story's top-level kids, and each item/<id>.json; it stores one job per comment, a short excerpt with emails and phones removed, and links the comment (Sources, 2026-10-31). The endpoint URL must be a .json URL."
++ },
"item_kind": {
"type": "string",
"pattern": "^[a-z][a-z0-9_]{1,39}$"
@@@ -160,8 +173,14 @@
"max_bytes": {
"type": "integer",
"minimum": 1024,
-- "maximum": 33554432,
-- "description": "The body cap for this endpoint, in bytes (1 KiB to 32 MiB); the deployment's cap otherwise. OVHcloud's VPS catalog is about 12 MB (Sources#51)."
++ "maximum": 17179869184,
++ "description": "The body cap for this endpoint, in bytes; the deployment's cap otherwise. An api endpoint may set 1 KiB to 64 MiB (OVHcloud's VPS catalog is about 12 MB; Sources#51); a bulk dataset file may set up to 16 GiB, and the row limit stops the stream long before that (Sources, 2026-10-31)."
++ },
++ "max_rows": {
++ "type": "integer",
++ "minimum": 1,
++ "maximum": 5000000,
++ "description": "A bulk endpoint only: the most rows one run keeps (default 1000). The stream is stopped once this many items are stored, so a multi-GB file is never held in memory and never read whole (Sources, 2026-10-31)."
},
"timeout_ms": {
"type": "integer",
ddiff --git a/seeds/sources.json b/seeds/sources.json
iindex 1a75dbf..877c8db 100644
---- a/seeds/sources.json
++++ b/seeds/sources.json
@@@ -2247,6 +2247,184 @@
"review_required": true,
"sensitivity": "none",
"default_indexability": "noindex"
++ },
++ {
++ "key": "bls-cpi-rent",
++ "name": "BLS CPI: rent of primary residence, US city average and 19 metro areas",
++ "type": "api",
++ "category": "housing",
++ "homepage_url": "https://www.bls.gov/cpi/",
++ "endpoints": [
++ {
++ "url": "https://download.bls.gov/pub/time.series/cu/cu.data.0.Current",
++ "format": "tsv",
++ "items_path": "",
++ "item_kind": "rent_cpi",
++ "max_bytes": 67108864,
++ "timeout_ms": 120000,
++ "filter": {
++ "series_id": [
++ "CUUR0000SEHA",
++ "CUURS11ASEHA",
++ "CUURS12ASEHA",
++ "CUURS12BSEHA",
++ "CUURS23ASEHA",
++ "CUURS23BSEHA",
++ "CUURS24ASEHA",
++ "CUURS35ASEHA",
++ "CUURS35BSEHA",
++ "CUURS35CSEHA",
++ "CUURS37ASEHA",
++ "CUURS37BSEHA",
++ "CUURS48ASEHA",
++ "CUURS48BSEHA",
++ "CUURS49ASEHA",
++ "CUURS49BSEHA",
++ "CUURS49DSEHA",
++ "CUURS49ESEHA",
++ "CUURA104SEHA",
++ "CUURA210SEHA"
++ ]
++ },
++ "fields": {
++ "identity": "$index",
++ "title": "series_id"
++ },
++ "extra": {
++ "series_id": "series_id",
++ "year": "year",
++ "period": "period",
++ "value": "value"
++ }
++ }
++ ],
++ "robots_note": "download.bls.gov/robots.txt answers 404 to the OpenVibeBot UA (re-checked 2026-10-30 from openvibe-ovh; an unmarked fetch gets 403, so the fetcher always sends the ADR 0007 User-Agent). api.bls.gov is Disallow: / and is not used. The fetcher re-checks robots.txt on every run.",
++ "terms_note": "BLS public data. BLS asks to be cited and says it cannot vouch for data or analyses derived from these data after they have been retrieved from BLS.gov (https://www.bls.gov/developers/termsOfService.htm, https://www.bls.gov/bls/linksite.htm). This is BLS's published CPI flat file (cu.data.0.Current, about 49 MB, every series oldest to newest): the filter keeps the US city average rent series CUUR0000SEHA and the 19 largest metro rent series, one item per month, so Rent can draw a dated national and metro trend line. Monthly data; a weekly poll is plenty. Store the series id, year, period and value; cite BLS and link https://data.bls.gov/timeseries/<series>. Re-verify before enabling.",
++ "license_note": "US federal government work (public domain in the US); BLS cited as the source.",
++ "min_interval_ms": 60000,
++ "poll_interval_sec": 604800,
++ "max_items": 12000,
++ "enabled": false,
++ "review_required": true,
++ "sensitivity": "none",
++ "default_indexability": "noindex"
++ },
++ {
++ "key": "open-food-facts",
++ "name": "Open Food Facts product database (daily export)",
++ "type": "bulk",
++ "category": "food",
++ "homepage_url": "https://world.openfoodfacts.org/",
++ "endpoints": [
++ {
++ "url": "https://static.openfoodfacts.org/data/en.openfoodfacts.org.products.csv.gz",
++ "format": "tsv",
++ "item_kind": "food_product",
++ "max_bytes": 2147483648,
++ "max_rows": 12000,
++ "timeout_ms": 120000,
++ "fields": {
++ "identity": "code",
++ "url": "url",
++ "title": "product_name"
++ },
++ "extra": {
++ "brands": "brands",
++ "categories": "categories",
++ "countries": "countries_en",
++ "main_category": "main_category_en",
++ "quantity": "quantity",
++ "nutriscore_grade": "nutriscore_grade",
++ "nutriscore_score": "nutriscore_score",
++ "energy_kcal_100g": "energy-kcal_100g",
++ "fat_100g": "fat_100g",
++ "saturated_fat_100g": "saturated-fat_100g",
++ "carbohydrates_100g": "carbohydrates_100g",
++ "sugars_100g": "sugars_100g",
++ "proteins_100g": "proteins_100g",
++ "salt_100g": "salt_100g",
++ "fiber_100g": "fiber_100g",
++ "sodium_100g": "sodium_100g"
++ }
++ }
++ ],
++ "robots_note": "static.openfoodfacts.org/robots.txt checked 2026-10-30 from openvibe-ovh: 'User-agent: *', 'Disallow: /api', 'Disallow: /cgi', 'Disallow: /facets'; /data/ is allowed. The download redirects to openfoodfacts-ds.s3.eu-west-3.amazonaws.com, whose /robots.txt is 404 (no restrictions). The fetcher re-checks robots.txt on both origins, and every redirect hop, on every run.",
++ "terms_note": "Open Food Facts publishes its database under the Open Database License: 'The Open Food Facts database is available under the Open Database License' (https://world.openfoodfacts.org/terms-of-use). Attribution and share-alike apply, and the ODbL notice stays with anything derived. This is the tab-separated CSV export (en.openfoodfacts.org.products.csv.gz, about 1.5 GB compressed): the bulk loader streams it with a 2 GiB cap and stops after 12,000 rows per run, so it never downloads the file whole. Store the code, name, brands, categories, Nutri-Score and the per-100 g nutrient columns; images are CC BY-SA and are not stored. Re-verify before enabling.",
++ "license_note": "ODbL 1.0 (database); individual contents DbCL; images CC BY-SA. Credit 'Open Food Facts (ODbL)' and link the product page and https://opendatacommons.org/licenses/odbl/1-0/ wherever a product's figures show.",
++ "min_interval_ms": 60000,
++ "poll_interval_sec": 86400,
++ "max_items": 12000,
++ "enabled": false,
++ "review_required": true,
++ "sensitivity": "none",
++ "default_indexability": "noindex"
++ },
++ {
++ "key": "themealdb",
++ "name": "TheMealDB recipes",
++ "type": "api",
++ "category": "food",
++ "homepage_url": "https://www.themealdb.com/",
++ "endpoints": [
++ {
++ "url": "https://www.themealdb.com/api/json/v1/1/search.php?s=",
++ "format": "json",
++ "items_path": "meals",
++ "item_kind": "recipe",
++ "max_bytes": 1048576,
++ "fields": {
++ "identity": "idMeal",
++ "title": "strMeal",
++ "summary": "strInstructions"
++ },
++ "extra": {
++ "category": "strCategory",
++ "area": "strArea",
++ "thumbnail": "strMealThumb",
++ "tags": "strTags",
++ "youtube": "strYoutube",
++ "source": "strSource"
++ }
++ }
++ ],
++ "robots_note": "www.themealdb.com/robots.txt checked 2026-10-30 from openvibe-ovh: 'User-agent: *' then 'Allow: /', and a Sitemap line; no Disallow. The fetcher re-checks it on every run.",
++ "terms_note": "TheMealDB says 'The API and site will always remain free at point of access' (https://www.themealdb.com/api.php). The key in this URL is the test key '1', which TheMealDB documents for development: a production key is the owner's call, so this seed stays disabled. Only a short plain-text excerpt of the instructions is stored (summary, capped at 1000 characters), the recipe is linked by its own URL, and TheMealDB is attributed; the recipe text and images remain the site's. Re-verify before enabling.",
++ "license_note": "TheMealDB's own free API terms; recipe text and images remain the site's. A short excerpt and a link back only.",
++ "min_interval_ms": 60000,
++ "poll_interval_sec": 604800,
++ "max_items": 400,
++ "enabled": false,
++ "review_required": true,
++ "sensitivity": "none",
++ "default_indexability": "noindex"
++ },
++ {
++ "key": "hn-whoishiring",
++ "name": "Hacker News: Who is hiring (monthly thread)",
++ "type": "api",
++ "category": "jobs",
++ "homepage_url": "https://news.ycombinator.com/",
++ "endpoints": [
++ {
++ "url": "https://hacker-news.firebaseio.com/v0/user/whoishiring.json",
++ "format": "json",
++ "resolver": "hn-whoishiring",
++ "item_kind": "job",
++ "max_bytes": 1048576,
++ "timeout_ms": 20000
++ }
++ ],
++ "robots_note": "hacker-news.firebaseio.com/robots.txt checked 2026-10-30 from openvibe-ovh: 'User-agent: *', 'Allow: /*.json$', 'Allow: /*.json?*', 'Disallow: /'. The resolver asks only for .json paths (user/whoishiring.json, item/<id>.json) and refuses any other path before fetching. The fetcher re-checks robots.txt on every run.",
++ "terms_note": "The Hacker News API makes 'the public Hacker News data available in near real time'; 'There is currently no rate limit' (https://github.com/HackerNews/API). The resolver walks user/whoishiring.json to the newest 'Ask HN: Who is hiring?' story and only its top-level kids, then item/<id>.json once per comment; it never walks a comment's replies. One job item per comment, with the comment linked: the full comment is never stored, only a plain-text excerpt with email addresses and phone numbers replaced by [email] and [phone]. The poster owns the comment. Attribute Hacker News and the poster; link https://news.ycombinator.com/item?id=<id>. Re-verify before enabling.",
++ "license_note": "Hacker News user content; the poster owns the comment. Headline, short excerpt and link only.",
++ "min_interval_ms": 1000,
++ "poll_interval_sec": 86400,
++ "max_items": 500,
++ "retire_after_sec": 3888000,
++ "enabled": false,
++ "review_required": true,
++ "sensitivity": "none",
++ "default_indexability": "noindex"
}
]
}
ddiff --git a/server/adapters/api.js b/server/adapters/api.js
iindex e13361d..17c223c 100644
---- a/server/adapters/api.js
++++ b/server/adapters/api.js
@@@ -28,12 +28,14 @@
*
* "extra": { "price_hourly": "$root.terms.OnDemand.{$key}.*.priceDimensions.*.pricePerUnit.USD" }
*
-- * An endpoint may carry `where`: an object of up to 10 `path -> expected` conditions whose paths
-- * follow the same rules (relative, `$key`, `$root.`, `*`). Expected is a string, number or boolean
-- * (compared to the value after `scalar()`), or an array of up to 20 such values (any of them
-- * matches). A record that fails any condition is left out before `maxItems` is counted and
-- * reported as `meta.filtered`, not `skipped`:
++ * An endpoint may carry `where` (or its alias `filter`): an object of up to 10 `path -> expected`
++ * conditions whose paths follow the same rules (relative, `$key`, `$root.`, `*`). Expected is a
++ * string, number or boolean (compared to the value after `scalar()`), or an array of up to 20 such
++ * values (any of them matches). A record that fails any condition is left out before `maxItems` is
++ * counted and reported as `meta.filtered`, not `skipped`. `filter` is the name the BLS CPI rent
++ * seed uses; both are applied when both are given:
*
++ * "filter": { "series_id": "CUUR0000SEHA" }
* "where": { "productFamily": "Lightsail Instance", "attributes.operatingSystem": "Linux" }
*
* A key may carry a selector, `key[field=value,…]` (one to four conditions): `key` must be an array, and
@@@ -49,7 +51,7 @@
* A `where` expectation may also be `{ "pattern": "<regular expression>" }` (at most 200 characters):
* the value must be text the expression matches, e.g. `{ "planCode": { "pattern": "^vps-[0-9]{4}-model[0-9]+$" } }`.
*
-- * `max_bytes` (an integer from 1 KiB to 32 MiB) raises or lowers the body cap for one endpoint
++ * `max_bytes` (an integer from 1 KiB to 64 MiB) raises or lowers the body cap for one endpoint
* (SOURCES_MAX_BYTES otherwise); OVHcloud's VPS catalog is about 12 MB.
*
* `timeout_ms` (an integer from 1 s to 120 s) gives one endpoint its own deadline (SOURCES_FETCH_TIMEOUT_MS otherwise):
@@@ -222,14 +224,37 @@ function keyPathViolation(p, shape) {
return null;
}
--/** Validate an endpoint mapping (registry side). Returns an error string or null. */
--function checkMapping(ep) {
-- if (!['json', 'xml', 'tsv'].includes(ep.format || 'json')) return 'format must be json, xml or tsv';
-- if ((ep.format || 'json') === 'tsv' && ep.items_path) return 'a tsv body is the array itself: items_path is ""';
-- if (ep.duplicates !== undefined && !['first', 'last'].includes(ep.duplicates)) return 'duplicates must be first or last';
++/** The `where` / `filter` conditions: path -> expected value(s) or { pattern }. */
++function checkConditions(where, label, shape) {
++ if (typeof where !== 'object' || Array.isArray(where)) return `${label} must be an object of path -> expected`;
++ if (Object.keys(where).length > 10) return `at most 10 ${label} conditions`;
++ for (const [p, expected] of Object.entries(where)) {
++ if (!PATH_RE.test(p)) return `${label}.${p} path is malformed`;
++ const kerr = keyPathViolation(p, shape);
++ if (kerr) return `${label}.${p}: ${kerr}`;
++ if (isPattern(expected)) {
++ if (Object.keys(expected).length !== 1 || typeof expected.pattern !== 'string' || !expected.pattern || expected.pattern.length > 200) return `${label}.${p}: a pattern is { "pattern": "<at most 200 characters>" }`;
++ try { new RegExp(expected.pattern); } catch { return `${label}.${p}: the pattern is not a valid regular expression`; }
++ continue;
++ }
++ const wanted = Array.isArray(expected) ? expected : [expected];
++ if (wanted.length > 20) return `${label}.${p} has at most 20 expected values`;
++ for (const v of wanted) {
++ if (typeof v !== 'string' && typeof v !== 'number' && typeof v !== 'boolean') return `${label}.${p} expectations must be strings, numbers or booleans`;
++ }
++ }
++ return null;
++}
++
++/** Validate an endpoint mapping (registry side). `bulk` is the streaming dataset download's own schema. */
++function checkMapping(ep, { bulk = false } = {}) {
++ const formats = bulk ? ['csv', 'jsonl', 'tsv'] : ['json', 'xml', 'tsv'];
++ if (!formats.includes(ep.format || (bulk ? 'csv' : 'json'))) return bulk ? 'format must be csv, jsonl or tsv' : 'format must be json, xml or tsv';
++ if (['tsv', 'csv', 'jsonl'].includes(ep.format) && ep.items_path) return `a ${ep.format} body is the array itself: items_path is ""`;
++ if (!bulk && ep.duplicates !== undefined && !['first', 'last'].includes(ep.duplicates)) return 'duplicates must be first or last';
const shape = ep.items_shape || 'array';
-- if (!['array', 'map'].includes(shape)) return 'items_shape must be array or map';
-- if (ep.items_path !== undefined && ep.items_path !== '' && !PATH_RE.test(ep.items_path)) return 'items_path is malformed';
++ if (!bulk && !['array', 'map'].includes(shape)) return 'items_shape must be array or map';
++ if (!bulk && ep.items_path !== undefined && ep.items_path !== '' && !PATH_RE.test(ep.items_path)) return 'items_path is malformed';
const fields = ep.fields || {};
if (!fields.identity && !fields.url) return 'fields.identity or fields.url is required';
for (const [k, p] of Object.entries(fields)) {
@@@ -254,31 +279,22 @@ function checkMapping(ep) {
const kerr = keyPathViolation(p, shape);
if (kerr) return `extra_lists.${k}: ${kerr}`;
}
-- const where = ep.where == null ? {} : ep.where;
-- if (typeof where !== 'object' || Array.isArray(where)) return 'where must be an object of path -> expected';
-- if (Object.keys(where).length > 10) return 'at most 10 where conditions';
-- for (const [p, expected] of Object.entries(where)) {
-- if (!PATH_RE.test(p)) return `where.${p} path is malformed`;
-- const kerr = keyPathViolation(p, shape);
-- if (kerr) return `where.${p}: ${kerr}`;
-- if (isPattern(expected)) {
-- if (Object.keys(expected).length !== 1 || typeof expected.pattern !== 'string' || !expected.pattern || expected.pattern.length > 200) return `where.${p}: a pattern is { "pattern": "<at most 200 characters>" }`;
-- try { new RegExp(expected.pattern); } catch { return `where.${p}: the pattern is not a valid regular expression`; }
-- continue;
-- }
-- const wanted = Array.isArray(expected) ? expected : [expected];
-- if (wanted.length > 20) return `where.${p} has at most 20 expected values`;
-- for (const v of wanted) {
-- if (typeof v !== 'string' && typeof v !== 'number' && typeof v !== 'boolean') return `where.${p} expectations must be strings, numbers or booleans`;
-- }
++ for (const label of ['where', 'filter']) {
++ if (ep[label] == null) continue;
++ const err = checkConditions(ep[label], label, shape);
++ if (err) return err;
}
if (ep.item_kind !== undefined && !/^[a-z][a-z0-9_]{1,39}$/.test(ep.item_kind)) return 'item_kind is malformed';
-- if (ep.max_bytes !== undefined && (!Number.isInteger(ep.max_bytes) || ep.max_bytes < 1024 || ep.max_bytes > 32 * 1024 * 1024)) return 'max_bytes is an integer from 1024 to 33554432';
++ if (ep.max_bytes !== undefined && (!Number.isInteger(ep.max_bytes) || ep.max_bytes < 1024 || ep.max_bytes > (bulk ? 16 * 1024 * 1024 * 1024 : 64 * 1024 * 1024))) {
++ return bulk ? 'max_bytes is an integer from 1024 to 17179869184' : 'max_bytes is an integer from 1024 to 67108864';
++ }
++ if (ep.max_rows !== undefined && (!bulk || !Number.isInteger(ep.max_rows) || ep.max_rows < 1 || ep.max_rows > 5000000)) return 'max_rows is an integer from 1 to 5000000';
if (ep.timeout_ms !== undefined && (!Number.isInteger(ep.timeout_ms) || ep.timeout_ms < 1000 || ep.timeout_ms > 120000)) return 'timeout_ms is an integer from 1000 to 120000';
-- if (ep.request_content_type !== undefined && ep.request_content_type !== 'application/json') return 'request_content_type may only be application/json';
++ if (!bulk && ep.request_content_type !== undefined && ep.request_content_type !== 'application/json') return 'request_content_type may only be application/json';
return null;
}
++
/** A tab-separated body as records keyed by its header line's names (values trimmed); malformed lines are counted. */
function parseTsv(text) {
const lines = String(text).split(/\r?\n/).filter((l) => l.trim() !== '');
@@@ -297,6 +313,29 @@ function parseTsv(text) {
return { records, bad };
}
++/** One record → the item every api-shaped adapter stores, or null when it states no identity. */
++function mapRecord(body, rec, key, index, endpoint, url, summaryMax = 1000) {
++ const f = endpoint.fields || {};
++ const get = (p) => (p === '$index' ? index : getField(body, rec, key, p));
++ const link = f.url ? canonicalUrl(scalar(get(f.url)), url) : null;
++ const identity = (f.identity ? toId(scalar(get(f.identity))) : null) || link;
++ if (!identity) return null;
++ const extra = {};
++ for (const [k, p] of Object.entries(endpoint.extra || {})) extra[k] = scalar(get(p));
++ for (const [k, p] of Object.entries(endpoint.extra_lists || {})) extra[k] = scalars(get(p));
++ return {
++ identity: String(identity),
++ kind: endpoint.item_kind || 'record',
++ canonical_url: link,
++ title: f.title ? toText(scalar(get(f.title)), 500) : null,
++ summary: f.summary ? toText(scalar(get(f.summary)), summaryMax) : null,
++ authors: f.author ? [toText(scalar(get(f.author)), 200)].filter(Boolean) : [],
++ published_at: f.published_at ? toIsoDate(scalar(get(f.published_at))) : f.published_at_epoch ? epochSecondsToIso(scalar(get(f.published_at_epoch))) : null,
++ updated_at: f.updated_at ? toIsoDate(scalar(get(f.updated_at))) : f.updated_at_epoch ? epochSecondsToIso(scalar(get(f.updated_at_epoch))) : null,
++ fields: extra,
++ };
++}
++
function parse(text, { url, endpoint = {}, summaryMax = 1000, maxItems = 500 } = {}) {
const format = endpoint.format || 'json';
let body;
@@@ -317,9 +356,11 @@ function parse(text, { url, endpoint = {}, summaryMax = 1000, maxItems = 500 } =
if (!Array.isArray(node)) throw new ParseError(`items_path "${itemsPath}" is not an array in the response`);
records = node.map((value, index) => ({ key: undefined, index, value }));
}
-- const f = endpoint.fields || {};
-- const where = endpoint.where && typeof endpoint.where === 'object' && !Array.isArray(endpoint.where) ? endpoint.where : {};
-- const filterKeys = Object.keys(where);
++ const conditions = [];
++ for (const label of ['filter', 'where']) {
++ const c = endpoint[label];
++ if (c && typeof c === 'object' && !Array.isArray(c)) conditions.push(c);
++ }
const items = [];
let skipped = malformed;
let filtered = 0;
@@@ -327,32 +368,16 @@ function parse(text, { url, endpoint = {}, summaryMax = 1000, maxItems = 500 } =
const keepLast = endpoint.duplicates === 'last';
for (const { key, index, value: rec } of records) {
if (!rec || typeof rec !== 'object' || (shape === 'map' && Array.isArray(rec))) { skipped++; continue; }
-- if (filterKeys.length && !matchesWhere(body, rec, key, where)) { filtered++; continue; }
-- const get = (p) => (p === '$index' ? index : getField(body, rec, key, p));
-- const link = f.url ? canonicalUrl(scalar(get(f.url)), url) : null;
-- const identity = (f.identity ? toId(scalar(get(f.identity))) : null) || link;
-- const again = identity && seen.has(identity);
-- if (!identity || (again && !keepLast)) { skipped++; continue; }
++ if (conditions.length && !conditions.every((c) => matchesWhere(body, rec, key, c))) { filtered++; continue; }
++ const item = mapRecord(body, rec, key, index, endpoint, url, summaryMax);
++ const again = item && seen.has(item.identity);
++ if (!item || (again && !keepLast)) { skipped++; continue; }
if (!again && items.length >= maxItems) { skipped++; continue; }
if (again) skipped++; // the earlier record it replaces
-- const extra = {};
-- for (const [k, p] of Object.entries(endpoint.extra || {})) extra[k] = scalar(get(p));
-- for (const [k, p] of Object.entries(endpoint.extra_lists || {})) extra[k] = scalars(get(p));
-- const item = {
-- identity: String(identity),
-- kind: endpoint.item_kind || 'record',
-- canonical_url: link,
-- title: f.title ? toText(scalar(get(f.title)), 500) : null,
-- summary: f.summary ? toText(scalar(get(f.summary)), summaryMax) : null,
-- authors: f.author ? [toText(scalar(get(f.author)), 200)].filter(Boolean) : [],
-- published_at: f.published_at ? toIsoDate(scalar(get(f.published_at))) : f.published_at_epoch ? epochSecondsToIso(scalar(get(f.published_at_epoch))) : null,
-- updated_at: f.updated_at ? toIsoDate(scalar(get(f.updated_at))) : f.updated_at_epoch ? epochSecondsToIso(scalar(get(f.updated_at_epoch))) : null,
-- fields: extra,
-- };
-- if (again) items[seen.get(identity)] = item;
-- else { seen.set(identity, items.length); items.push(item); }
++ if (again) items[seen.get(item.identity)] = item;
++ else { seen.set(item.identity, items.length); items.push(item); }
}
return { items, skipped, meta: { format, filtered } };
}
--module.exports = { parse, PARSER_VERSION, checkMapping };
++module.exports = { parse, mapRecord, PARSER_VERSION, checkMapping };
ddiff --git a/server/adapters/bulk.js b/server/adapters/bulk.js
nnew file mode 100644
iindex 0000000..9e6a8a4
---- /dev/null
++++ b/server/adapters/bulk.js
@@@ -0,0 +1,162 @@
++'use strict';
++/**
++ * Bulk datasets: one dataset file per source, read as a stream, never loaded whole.
++ *
++ * { "url": "https://static.example.org/products.csv.gz", "format": "csv",
++ * "item_kind": "product", "max_bytes": 2147483648, "max_rows": 12000,
++ * "fields": { "identity": "code", "url": "url", "title": "product_name" },
++ * "extra": { "brands": "brands", "energy_kcal": "energy-kcal_100g" } }
++ *
++ * The seed names the format — CSV, JSONL or TSV — and the field paths are the same as the api adapter's (a
++ * column name for CSV/TSV, a dotted path inside the JSON object for JSONL). The parser takes chunks from the
++ * fetcher and emits items as it goes; it stops the stream as soon as `max_rows` items are stored, so a 13 GB
++ * dump costs a few MB of transfer and an ordinary amount of memory. `max_bytes` is the stream's own size cap.
++ *
++ * CSV follows RFC 4180 enough for published dumps: quoted fields, a quote inside a quoted field ("") and line
++ * breaks inside quotes. TSV reads its header line the way api@1's tsv format does. JSONL is one JSON object per
++ * non-empty line. A malformed row is counted and skipped, never invented.
++ */
++const { StringDecoder } = require('string_decoder');
++const { mapRecord } = require('./api');
++const { ParseError } = require('./util');
++
++const PARSER_VERSION = 'bulk@1';
++
++function createParser({ endpoint = {}, url, maxItems = 500, summaryMax = 1000 } = {}) {
++ const format = endpoint.format || 'csv';
++ const limit = Math.max(1, Math.min(Number(endpoint.max_rows) || 1000, Number(maxItems) || 500));
++ const decoder = new StringDecoder('utf8');
++ const items = [];
++ const seen = new Set();
++ let skipped = 0;
++ let rows = 0;
++ let stopped = false;
++
++ // Line-oriented formats keep the tail of an incomplete line between chunks. CSV needs a small state machine
++ // because a quoted field may contain commas and newlines.
++ let pending = '';
++ let header = null;
++ let field = '';
++ let row = [];
++ let inQuotes = false;
++ let quotePending = false;
++ let first = true;
++
++ function accept(rec) {
++ if (stopped) return;
++ if (!rec || typeof rec !== 'object' || Array.isArray(rec)) { skipped++; return; }
++ const index = rows++;
++ const item = mapRecord(rec, rec, undefined, index, endpoint, url, summaryMax);
++ if (!item) { skipped++; return; }
++ if (seen.has(item.identity)) { skipped++; return; }
++ seen.add(item.identity);
++ items.push(item);
++ if (items.length >= limit) stopped = true;
++ }
++
++ function headerOf(names, what) {
++ const clean = names.map((n) => String(n == null ? '' : n).replace(/^\uFEFF/, '').trim());
++ if (clean.some((n) => !n) || new Set(clean).size !== clean.length) throw new ParseError(`${what} header names must be present and distinct`);
++ return clean;
++ }
++
++ function csvRow(cells) {
++ if (first) {
++ first = false;
++ if (cells.length === 1 && cells[0].trim() === '') return; // a blank first line
++ header = headerOf(cells, 'csv');
++ return;
++ }
++ if (cells.length === 1 && cells[0].trim() === '') return;
++ if (cells.length !== header.length) { skipped++; return; }
++ const rec = {};
++ header.forEach((n, i) => { rec[n] = String(cells[i]).trim(); });
++ accept(rec);
++ }
++
++ function csvWrite(text) {
++ for (let i = 0; i < text.length; i++) {
++ if (stopped) return;
++ const ch = text[i];
++ if (quotePending) {
++ quotePending = false;
++ if (ch === '"') { field += '"'; continue; }
++ inQuotes = false;
++ i--; // the close-quote's next character is an ordinary one
++ continue;
++ }
++ if (inQuotes) {
++ if (ch === '"') quotePending = true;
++ else field += ch;
++ continue;
++ }
++ if (ch === '"' && field === '') { inQuotes = true; continue; }
++ if (ch === ',') { row.push(field); field = ''; continue; }
++ if (ch === '\n') { row.push(field); field = ''; const cells = row; row = []; csvRow(cells); continue; }
++ if (ch === '\r') continue;
++ field += ch;
++ }
++ }
++
++ function csvFinish() {
++ if (quotePending) { quotePending = false; inQuotes = false; }
++ if (inQuotes) { skipped++; return; } // an unterminated quoted field is not a row
++ if (!field.length && !row.length) return;
++ row.push(field);
++ field = '';
++ const cells = row;
++ row = [];
++ csvRow(cells);
++ }
++
++ function line(text) {
++ if (text === '') return;
++ if (format === 'jsonl') {
++ let rec;
++ try { rec = JSON.parse(text); } catch { skipped++; return; }
++ accept(rec);
++ return;
++ }
++ // tsv
++ const cells = text.split('\t');
++ if (first) { first = false; header = headerOf(cells, 'tsv'); return; }
++ if (cells.length !== header.length) { skipped++; return; }
++ const rec = {};
++ header.forEach((n, i) => { rec[n] = String(cells[i]).trim(); });
++ accept(rec);
++ }
++
++ function lineWrite(text) {
++ pending += text;
++ const parts = pending.split('\n');
++ pending = parts.pop();
++ for (const part of parts) {
++ line(part.endsWith('\r') ? part.slice(0, -1) : part);
++ if (stopped) return;
++ }
++ }
++
++ /** Feed one chunk; returns false when the row limit is reached and the fetcher should stop. */
++ function write(chunk) {
++ if (stopped) return false;
++ const text = decoder.write(chunk);
++ if (!text) return !stopped;
++ if (format === 'csv') csvWrite(text);
++ else lineWrite(text);
++ return !stopped;
++ }
++
++ /** Finish the stream (not called when the row limit stopped it): flush the last line/row. */
++ function end() {
++ if (!stopped) {
++ const tail = decoder.end();
++ if (format === 'csv') { if (tail) csvWrite(tail); csvFinish(); }
++ else { if (tail) lineWrite(tail); if (pending) line(pending.endsWith('\r') ? pending.slice(0, -1) : pending); }
++ }
++ return { items, skipped, meta: { format, rows, filtered: 0, capped: stopped } };
++ }
++
++ return { write, end, get full() { return stopped; }, get count() { return items.length; } };
++}
++
++module.exports = { createParser, PARSER_VERSION };
ddiff --git a/server/adapters/hn.js b/server/adapters/hn.js
nnew file mode 100644
iindex 0000000..e9be690
---- /dev/null
++++ b/server/adapters/hn.js
@@@ -0,0 +1,119 @@
++'use strict';
++/**
++ * Hacker News "Who is hiring": a resolver, not one items_path. The monthly thread lives across several JSON
++ * documents, so this walks the same way a reader would:
++ *
++ * user/whoishiring.json → the newest story whose title is "Ask HN: Who is hiring?" → its top-level
++ * kids → item/<id>.json for each comment
++ *
++ * Only the top-level comments are the job posts; a comment's own kids are never fetched, so a thread of a few
++ * hundred posts costs a few hundred requests and no tree walk. One `job` item per comment, with the comment's
++ * canonical URL — the full comment is only linked, never stored. The stored excerpt has email addresses and
++ * phone numbers replaced with `[email]` and `[phone]`; the title is the first line of the comment, stripped the
++ * same way. The API's robots.txt allows only `/*.json` (`Allow: /*.json$`, `Disallow: /`), so every URL this
++ * resolver asks for is checked to end in `.json` before it is fetched.
++ *
++ * A test passes recorded fixtures as `getJson`; nothing here reaches the network by itself.
++ */
++const { ParseError, toText } = require('./util');
++
++const PARSER_VERSION = 'hn-whoishiring@1';
++const STORY_RE = /^Ask HN:\s*Who is hiring\?/i;
++const COMMENTS_URL = 'https://news.ycombinator.com/item?id=';
++
++/** The API's robots rule: only a JSON path is allowed. */
++function assertJsonUrl(value) {
++ let u;
++ try { u = new URL(String(value)); } catch { throw new ParseError(`not a URL: ${value}`); }
++ if (!u.pathname.endsWith('.json')) throw new ParseError(`the Hacker News API allows only .json paths, not ${u.pathname}`);
++ return u.toString();
++}
++
++/** HTML entities and tags → the plain text a comment reads as. */
++function htmlToText(html) {
++ let s = String(html == null ? '' : html);
++ s = s.replace(/<\s*(?:p|br)\s*\/?\s*>/gi, '\n');
++ s = s.replace(/<[^>]*>/g, '');
++ s = s.replace(/&#x([0-9a-f]+);/gi, (_, h) => String.fromCodePoint(parseInt(h, 16)))
++ .replace(/&#(\d+);/g, (_, d) => String.fromCodePoint(Number(d)))
++ .replace(/&nbsp;/gi, ' ')
++ .replace(/&quot;/gi, '"')
++ .replace(/&#x27;|&#39;|&apos;/gi, "'")
++ .replace(/&lt;/gi, '<')
++ .replace(/&gt;/gi, '>')
++ .replace(/&amp;/gi, '&');
++ return s.split('\n').map((l) => l.replace(/[ \t]+/g, ' ').trim()).join('\n').replace(/\n{3,}/g, '\n\n').trim();
++}
++
++const EMAIL_RE = /[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}/gi;
++// A 10-digit North-American number with optional separators, an international +number, or an extension.
++const PHONE_RE = /(?:\+?\d{1,3}[\s.-]?)?(?:\(\d{3}\)|\d{3})[\s.-]?\d{3}[\s.-]?\d{4}(?:\s?(?:x|ext\.?)\s?\d{1,6})?|\+\d[\d\s().-]{7,}\d/gi;
++
++/** The excerpt's privacy rule: an email or a phone number is not stored. */
++function stripPrivate(value) {
++ return String(value == null ? '' : value).replace(EMAIL_RE, '[email]').replace(PHONE_RE, '[phone]');
++}
++
++/** A plain-text comment → the short stored excerpt (newlines become spaces, privacy applied before the cut). */
++function excerptOf(text, max) {
++ return stripPrivate(text).replace(/\s+/g, ' ').trim().slice(0, max);
++}
++
++/** The first line of a comment, which a Who is hiring post usually uses for company, role and place. */
++function titleOf(text) {
++ const line = stripPrivate(text).split('\n').map((l) => l.trim()).find(Boolean) || '';
++ return line.replace(/^[-–—*•\s]+/, '').slice(0, 200) || null;
++}
++
++async function resolve({ endpoint = {}, getJson, maxItems = 500, summaryMax = 1000, cap = 1_000_000 } = {}) {
++ if (typeof getJson !== 'function') throw new TypeError('resolve needs getJson');
++ const endpointUrl = assertJsonUrl(endpoint.url);
++ const u = new URL(endpointUrl);
++ const root = new URL('../', u); // …/v0/ from …/v0/user/whoishiring.json
++ const userUrl = new URL('user/whoishiring.json', root).toString();
++ const json = async (url) => {
++ const clean = assertJsonUrl(url);
++ const out = await getJson(clean, cap);
++ if (out == null || typeof out !== 'object') throw new ParseError(`Hacker News answered ${clean} with no object`);
++ return out;
++ };
++
++ const user = await json(userUrl);
++ const submitted = Array.isArray(user.submitted) ? user.submitted : [];
++ let story = null;
++ for (const id of submitted.slice(0, 12)) {
++ const item = await json(new URL(`item/${id}.json`, root).toString());
++ if (item && typeof item.title === 'string' && STORY_RE.test(item.title)) { story = item; break; }
++ }
++ if (!story) throw new ParseError('user/whoishiring.json has no "Ask HN: Who is hiring?" story');
++ const storyUrl = `${COMMENTS_URL}${story.id}`;
++
++ const kids = Array.isArray(story.kids) ? story.kids : [];
++ const items = [];
++ let skipped = 0;
++ for (const id of kids) {
++ if (items.length >= maxItems) { skipped += 1; continue; }
++ const comment = await json(new URL(`item/${id}.json`, root).toString());
++ if (!comment || comment.deleted || comment.dead || typeof comment.text !== 'string') { skipped++; continue; }
++ const text = htmlToText(comment.text);
++ if (!text) { skipped++; continue; }
++ const by = toText(comment.by, 200);
++ items.push({
++ identity: `hn-${comment.id}`,
++ kind: 'job',
++ canonical_url: `${COMMENTS_URL}${comment.id}`,
++ title: titleOf(text),
++ summary: excerptOf(text, summaryMax),
++ authors: by ? [by] : [],
++ published_at: Number.isFinite(Number(comment.time)) ? new Date(Number(comment.time) * 1000).toISOString() : null,
++ updated_at: null,
++ fields: { story_id: String(story.id), story_title: toText(story.title, 300), story_url: storyUrl },
++ });
++ }
++ return {
++ items, skipped,
++ meta: { format: 'json', filtered: 0, story_id: String(story.id), story_title: story.title, story_url: storyUrl, comments: kids.length },
++ };
++}
++
++module.exports = { resolve, stripPrivate, htmlToText, excerptOf, titleOf, assertJsonUrl, PARSER_VERSION, STORY_RE };
ddiff --git a/server/adapters/index.js b/server/adapters/index.js
iindex f1fc042..c29f29c 100644
---- a/server/adapters/index.js
++++ b/server/adapters/index.js
@@@ -1,12 +1,15 @@
'use strict';
/**
* Adapter registry: source type → parser. `manual` sources have no fetcher; their items are
-- * entered through the API with the evidence URL they came from.
++ * entered through the API with the evidence URL they came from. `bulk` streams one dataset file and
++ * `hn-whoishiring` walks several JSON documents (a resolver), so both have their own ingest path.
*/
const feed = require('./feed');
const sitemap = require('./sitemap');
const jsonld = require('./jsonld');
const api = require('./api');
++const bulk = require('./bulk');
++const hn = require('./hn');
const ADAPTERS = {
rss: { parse: feed.parse, version: feed.PARSER_VERSION, accept: 'application/rss+xml, application/atom+xml, application/xml;q=0.9, text/xml;q=0.8, */*;q=0.1' },
@@@ -14,8 +17,15 @@ const ADAPTERS = {
sitemap: { parse: sitemap.parse, version: sitemap.PARSER_VERSION, accept: 'application/xml, text/xml;q=0.9, */*;q=0.1' },
jsonld: { parse: jsonld.parse, version: jsonld.PARSER_VERSION, accept: 'text/html, application/xhtml+xml;q=0.9, */*;q=0.1' },
api: { parse: api.parse, version: api.PARSER_VERSION, accept: 'application/json, application/xml;q=0.9, text/tab-separated-values;q=0.9, text/plain;q=0.8, */*;q=0.1' },
++ bulk: {
++ parse: null, version: bulk.PARSER_VERSION, createParser: bulk.createParser,
++ accept: 'text/csv, text/tab-separated-values, application/x-ndjson, application/jsonl, application/json;q=0.9, application/gzip;q=0.9, */*;q=0.1',
++ },
};
++// A resolver reads several URLs of one API and returns the same { items, skipped, meta } an adapter's parse does.
++const RESOLVERS = { 'hn-whoishiring': { resolve: hn.resolve, version: hn.PARSER_VERSION } };
++
const MANUAL_VERSION = 'manual@1';
--module.exports = { ADAPTERS, MANUAL_VERSION, checkApiMapping: api.checkMapping };
++module.exports = { ADAPTERS, RESOLVERS, MANUAL_VERSION, checkApiMapping: api.checkMapping };
ddiff --git a/server/ingest.js b/server/ingest.js
iindex 29df1d2..1c24697 100644
---- a/server/ingest.js
++++ b/server/ingest.js
@@@ -25,11 +25,12 @@
* 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 crypto = require('crypto');
const { ids } = require('openvibe-contracts');
--const { ADAPTERS } = require('./adapters');
++const { ADAPTERS, RESOLVERS } = require('./adapters');
const extract = require('./adapters/extract');
const { createCrawler } = require('./crawl');
--const { sha256, decodeBody } = require('./adapters/util');
++const { sha256, decodeBody, ParseError } = require('./adapters/util');
const FAILURE_STATES = new Set(['http_error', 'timeout', 'robots_denied', 'parse_error', 'rate_limited']);
@@@ -229,6 +230,188 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
return { run, failed: false, success: true, recorded: true };
}
++ /**
++ * A bulk dataset file: one streamed GET, parsed row by row. The stream is stopped at the row limit, so a
++ * multi-GB file is never downloaded whole and never held in memory; the endpoint's max_bytes is the cap.
++ */
++ async function fetchBulkEndpoint(source, endpoint, trigger) {
++ const adapter = ADAPTERS.bulk;
++ const startedAt = now();
++ const run = baseRun(source, endpoint.url, trigger, startedAt);
++ const target = new URL(endpoint.url);
++ const host = target.host;
++
++ const headers = { Accept: adapter.accept };
++ const sensitive = [];
++ let requestUrl = target;
++ if (source.auth.mode !== 'none') {
++ const secret = config.secrets[source.auth.env];
++ if (!secret) {
++ Object.assign(run, { state: 'disabled', error_code: 'credential_missing', detail: `environment variable ${source.auth.env} is not set`, finished_at: now() });
++ return { run, failed: false };
++ }
++ if (source.auth.mode === 'bearer') { headers.Authorization = `Bearer ${secret}`; sensitive.push('Authorization'); }
++ if (source.auth.mode === 'header') { headers[source.auth.header] = secret; sensitive.push(source.auth.header); }
++ if (source.auth.mode === 'query') { requestUrl = new URL(target); requestUrl.searchParams.set(source.auth.param, secret); }
++ }
++
++ const verdict = await robots.check(target.toString());
++ setCrawlDelay(host, verdict.crawlDelaySec);
++ if (!verdict.allowed) {
++ if (verdict.refusedCode) Object.assign(run, { state: 'http_error', error_code: verdict.refusedCode, detail: `${host} is not fetchable (${verdict.refusedCode})`, finished_at: now() });
++ else Object.assign(run, { state: 'robots_denied', error_code: 'robots', detail: verdict.reason, finished_at: now() });
++ return { run, failed: true };
++ }
++
++ await space(host, source.min_interval_ms);
++ await st.requested.run(now(), source.key);
++ const parser = adapter.createParser({ endpoint, url: endpoint.url, maxItems: source.max_items, summaryMax: config.items.summaryMax });
++ const hash = crypto.createHash('sha256');
++ let res;
++ try {
++ res = await fetcher.streamUrl(requestUrl.toString(), {
++ headers,
++ capBytes: endpoint.max_bytes || config.fetch.maxBytes,
++ timeoutMs: endpoint.timeout_ms || undefined,
++ beforeHop: async (next) => {
++ const v = await robots.check(next.toString());
++ if (!v.allowed) return { refuse: v.reason };
++ if (next.host !== host) {
++ setCrawlDelay(next.host, v.crawlDelaySec);
++ await space(next.host, source.min_interval_ms);
++ }
++ return null;
++ },
++ }, (chunk) => { hash.update(chunk); return parser.write(chunk); });
++ } catch (err) {
++ const state = err.code === 'timeout' ? 'timeout' : err.code === 'hop_refused' ? 'robots_denied'
++ : err.code === 'parse_error' ? 'parse_error' : 'http_error';
++ Object.assign(run, { state, error_code: err.code || 'network', detail: String(err.message).slice(0, 500), finished_at: now() });
++ return { run, failed: true };
++ }
++ run.http_status = res.status;
++ run.bytes = res.bytes != null ? res.bytes : null;
++ run.finished_at = now();
++
++ if (res.status === 429) {
++ const wait = Math.min(retryAfterMs(res.headers['retry-after'], now()) ?? 10 * 60 * 1000, 24 * 3600 * 1000);
++ Object.assign(run, { state: 'rate_limited', error_code: 'upstream_429', detail: `the site asked us to wait ${Math.round(wait / 1000)} s` });
++ return { run, failed: true, notBefore: now() + wait };
++ }
++ if (res.status !== 200 && res.status !== 203) {
++ const wait = res.status === 503 ? retryAfterMs(res.headers['retry-after'], now()) : null;
++ Object.assign(run, { state: 'http_error', error_code: `http_${res.status}`, detail: `HTTP ${res.status}` });
++ return { run, failed: true, notBefore: wait != null ? now() + Math.min(wait, 24 * 3600 * 1000) : null };
++ }
++
++ let parsed;
++ try {
++ parsed = parser.end();
++ } catch (err) {
++ Object.assign(run, { state: 'parse_error', error_code: err.code === 'parse_error' ? 'unreadable' : 'adapter_error', detail: String(err.message).slice(0, 500) });
++ return { run, failed: true };
++ }
++ run.raw_body_hash = hash.digest('hex');
++ run.parser_version = adapter.version;
++ await db.tx(async () => {
++ const counts = await items.ingest(source, parsed.items, {
++ runId: run.id, rawBodyHash: run.raw_body_hash, parserVersion: adapter.version, retrievedAt: run.finished_at,
++ });
++ Object.assign(run, {
++ state: 'ok', items_seen: parsed.items.length, items_created: counts.created, items_updated: counts.updated,
++ items_unchanged: counts.unchanged, items_skipped: parsed.skipped + counts.skipped,
++ });
++ await st.insertRun.run(run);
++ await st.saveValidators.run({
++ k: source.key, u: endpoint.url, etag: res.headers.etag || null, lm: res.headers['last-modified'] || null,
++ status: res.status, at: run.finished_at,
++ });
++ });
++ return { run, failed: false, success: true, recorded: true };
++ }
++
++ /**
++ * An api endpoint with a resolver (HN Who is hiring): a few JSON documents walked by server/adapters/hn.js.
++ * Every URL is checked against robots and must be a .json path; the resolver returns the same shape a parser does.
++ */
++ async function fetchResolverEndpoint(source, endpoint, trigger) {
++ const adapter = ADAPTERS.api;
++ const { resolve, version: resolverVersion } = RESOLVERS[endpoint.resolver];
++ const startedAt = now();
++ const run = baseRun(source, endpoint.url, trigger, startedAt);
++ const target = new URL(endpoint.url);
++ const host = target.host;
++
++ const headers = { Accept: 'application/json' };
++ const sensitive = [];
++ let requestUrl = target;
++ if (source.auth.mode !== 'none') {
++ const secret = config.secrets[source.auth.env];
++ if (!secret) {
++ Object.assign(run, { state: 'disabled', error_code: 'credential_missing', detail: `environment variable ${source.auth.env} is not set`, finished_at: now() });
++ return { run, failed: false };
++ }
++ if (source.auth.mode === 'bearer') { headers.Authorization = `Bearer ${secret}`; sensitive.push('Authorization'); }
++ if (source.auth.mode === 'header') { headers[source.auth.header] = secret; sensitive.push(source.auth.header); }
++ if (source.auth.mode === 'query') { requestUrl = new URL(target); requestUrl.searchParams.set(source.auth.param, secret); }
++ }
++
++ const hash = crypto.createHash('sha256');
++ let bytes = 0;
++ let lastStatus = null;
++ const getJson = async (url, cap) => {
++ const verdict = await robots.check(url);
++ if (!verdict.allowed) throw Object.assign(new Error(verdict.reason || 'robots.txt disallows this URL'), { code: 'robots_denied' });
++ const r = await fetcher.fetchUrl(url, {
++ headers,
++ capBytes: Math.min(Number(cap) || 1048576, 32 * 1024 * 1024),
++ timeoutMs: endpoint.timeout_ms || undefined,
++ sensitiveHeaders: sensitive,
++ beforeHop: async (next) => {
++ const v = await robots.check(next.toString());
++ return v.allowed ? null : { refuse: v.reason };
++ },
++ });
++ lastStatus = r.status;
++ run.http_status = r.status;
++ if (r.status === 429) throw Object.assign(new Error('the Hacker News API answered 429'), { code: 'rate_limited' });
++ if (r.status !== 200 && r.status !== 203) throw Object.assign(new Error(`HTTP ${r.status} for ${url}`), { code: 'http_error' });
++ bytes += r.bytes || 0;
++ hash.update(r.body || Buffer.alloc(0));
++ try { return JSON.parse(decodeBody(r.body, r.headers['content-type'])); } catch { throw new ParseError('the Hacker News API answered non-JSON'); }
++ };
++
++ await space(host, source.min_interval_ms);
++ await st.requested.run(now(), source.key);
++ let parsed;
++ try {
++ parsed = await resolve({ endpoint, getJson, maxItems: source.max_items, summaryMax: config.items.summaryMax, cap: endpoint.max_bytes || 1048576 });
++ } catch (err) {
++ const state = err.code === 'robots_denied' ? 'robots_denied' : err.code === 'rate_limited' ? 'rate_limited'
++ : err.code === 'parse_error' ? 'parse_error' : err.code === 'timeout' ? 'timeout' : 'http_error';
++ Object.assign(run, { state, error_code: err.code || 'network', detail: String(err.message).slice(0, 500), finished_at: now() });
++ if (err.code !== 'parse_error') log.error(`[ingest] ${source.key}: resolver failed: ${err.stack || err}`);
++ return { run, failed: true };
++ }
++ run.http_status = lastStatus;
++ run.bytes = bytes;
++ run.finished_at = now();
++ run.raw_body_hash = hash.digest('hex');
++ run.parser_version = resolverVersion;
++ await db.tx(async () => {
++ const counts = await items.ingest(source, parsed.items, {
++ runId: run.id, rawBodyHash: run.raw_body_hash, parserVersion: resolverVersion, retrievedAt: run.finished_at,
++ });
++ Object.assign(run, {
++ state: 'ok', items_seen: parsed.items.length, items_created: counts.created, items_updated: counts.updated,
++ items_unchanged: counts.unchanged, items_skipped: parsed.skipped + counts.skipped,
++ });
++ await st.insertRun.run(run);
++ await st.saveValidators.run({ k: source.key, u: endpoint.url, etag: null, lm: null, status: lastStatus, at: run.finished_at });
++ });
++ 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) },
@@@ -245,7 +428,11 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
for (const endpoint of source.endpoints) {
let out;
try {
-- out = await fetchEndpoint(source, endpoint, trigger);
++ out = source.type === 'bulk'
++ ? await fetchBulkEndpoint(source, endpoint, trigger)
++ : endpoint.resolver
++ ? await fetchResolverEndpoint(source, endpoint, trigger)
++ : 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);
@@@ -262,7 +449,7 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
if (out.run.state === 'rate_limited') break; // the site asked us to stop
}
-- if (source.retire_after_sec && anySuccess && allAnswered && !anyFailure) {
++ if (source.retire_after_sec && source.type !== 'bulk' && 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`);
ddiff --git a/server/net/fetcher.js b/server/net/fetcher.js
iindex 89c4b13..c2c4683 100644
---- a/server/net/fetcher.js
++++ b/server/net/fetcher.js
@@@ -113,7 +113,113 @@ function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 102
}
}
-- return { fetchUrl };
++ /** A streamed GET: the same guard/redirect/deadline/cap, but the decoded bytes go to onChunk as they arrive. */
++ function onceStream(url, headers, deadline, capBytes, limitMs, onChunk) {
++ return new Promise((resolve, reject) => {
++ const mod = url.protocol === 'https:' ? https : http;
++ const remaining = deadline - Date.now();
++ if (remaining <= 0) return reject(new FetchError('timeout', 'deadline passed'));
++ let settled = false;
++ let stopped = false;
++ const done = (fn, v) => { if (!settled) { settled = true; clearTimeout(timer); fn(v); } };
++ const fail = (err) => { if (!settled) { settled = true; clearTimeout(timer); req.destroy(); reject(err); } };
++ const req = mod.request(url, {
++ method: 'GET',
++ headers,
++ lookup: guard.lookup,
++ agent: false,
++ timeout: remaining,
++ }, (res) => {
++ const status = res.statusCode;
++ if (status >= 300 && status < 400 && res.headers.location) {
++ res.resume();
++ return done(resolve, { redirect: res.headers.location, status, headers: res.headers });
++ }
++ // Non-200 answers (304, 429, 4xx/5xx) carry no dataset: read the headers and let ingest decide.
++ if (status !== 200 && status !== 203) {
++ res.resume();
++ return done(resolve, { status, headers: res.headers, bytes: 0, finalUrl: url.toString(), stopped: false });
++ }
++ const declared = Number(res.headers['content-length']);
++ if (Number.isFinite(declared) && declared > capBytes) {
++ res.resume();
++ return fail(new FetchError('too_large', `content-length ${declared} exceeds ${capBytes} bytes`));
++ }
++ let stream = res;
++ const enc = String(res.headers['content-encoding'] || '').trim().toLowerCase();
++ const type = String(res.headers['content-type'] || '').toLowerCase();
++ if (enc === 'gzip' || enc === 'x-gzip') stream = res.pipe(zlib.createGunzip());
++ else if (enc === 'deflate') stream = res.pipe(zlib.createInflate());
++ else if (enc === 'br') stream = res.pipe(zlib.createBrotliDecompress());
++ // A dataset file may be served as application/gzip (or named .gz) without Content-Encoding.
++ else if (/gzip/.test(type) || /\.gz(?:\?|$)/i.test(url.pathname + url.search)) stream = res.pipe(zlib.createGunzip());
++ let bytes = 0;
++ stream.on('data', (chunk) => {
++ if (settled) return;
++ bytes += chunk.length;
++ if (bytes > capBytes) return fail(new FetchError('too_large', `body exceeds ${capBytes} bytes`));
++ let more = true;
++ try { more = onChunk(chunk); } catch (err) { return fail(err); }
++ if (more === false) {
++ stopped = true;
++ done(resolve, { status, headers: res.headers, bytes, finalUrl: url.toString(), stopped: true });
++ res.destroy();
++ stream.destroy();
++ req.destroy();
++ }
++ });
++ stream.on('end', () => done(resolve, { status, headers: res.headers, bytes, finalUrl: url.toString(), stopped }));
++ stream.on('error', (err) => fail(new FetchError('network', `body: ${err.message}`)));
++ res.on('aborted', () => { if (!stopped) fail(new FetchError('network', 'response aborted')); });
++ });
++ const timer = setTimeout(() => {
++ req.destroy();
++ done(reject, new FetchError('timeout', `no complete response within ${limitMs} ms`));
++ }, remaining);
++ req.on('timeout', () => {
++ req.destroy();
++ done(reject, new FetchError('timeout', `no complete response within ${limitMs} ms`));
++ });
++ req.on('error', (err) => {
++ if (err instanceof GuardError) return done(reject, new FetchError(err.code, err.message));
++ done(reject, new FetchError('network', `${err.code || 'error'}: ${err.message}`));
++ });
++ req.end();
++ });
++ }
++
++ /**
++ * Stream one body to `onChunk(buffer)`. Return false from onChunk to stop early (the row limit): the request
++ * is destroyed and the promise resolves with `stopped: true`, so a multi-GB file is never downloaded whole.
++ */
++ async function streamUrl(input, { headers = {}, beforeHop = null, capBytes = maxBytes, timeoutMs: ownTimeout = null } = {}, onChunk) {
++ if (typeof onChunk !== 'function') throw new TypeError('streamUrl needs onChunk');
++ const limitMs = ownTimeout || timeoutMs;
++ const deadline = Date.now() + limitMs;
++ let url;
++ try { url = guard.checkUrl(input); } catch (err) { throw new FetchError(err.code || 'bad_url', err.message); }
++ const h = {
++ 'User-Agent': userAgent,
++ Accept: headers.Accept || '*/*',
++ 'Accept-Encoding': 'gzip, deflate, br',
++ ...headers,
++ };
++ for (let hop = 0; ; hop++) {
++ const r = await onceStream(url, h, deadline, capBytes, limitMs, onChunk);
++ if (!r.redirect) return { kind: 'response', status: r.status, headers: r.headers, body: null, bytes: r.bytes, stopped: r.stopped, finalUrl: url.toString() };
++ if (hop >= maxRedirects) throw new FetchError('too_many_redirects', `more than ${maxRedirects} redirects`);
++ let next;
++ try { next = guard.checkUrl(new URL(r.redirect, url).toString()); } catch (err) { throw new FetchError(err.code || 'bad_url', `redirect: ${err.message}`); }
++ if (beforeHop) {
++ const verdict = await beforeHop(next);
++ if (verdict && verdict.refuse) throw new FetchError('hop_refused', verdict.refuse);
++ }
++ if (next.origin !== url.origin) for (const name of ['Authorization']) delete h[name];
++ url = next;
++ }
++ }
++
++ return { fetchUrl, streamUrl };
}
module.exports = { createFetcher, FetchError };
ddiff --git a/server/registry.js b/server/registry.js
iindex aa24546..7418701 100644
---- a/server/registry.js
++++ b/server/registry.js
@@@ -13,7 +13,7 @@
const { checkApiMapping } = require('./adapters');
const extract = require('./adapters/extract');
--const TYPES = ['rss', 'atom', 'sitemap', 'jsonld', 'api', 'crawl', 'manual'];
++const TYPES = ['rss', 'atom', 'sitemap', 'jsonld', 'api', 'bulk', '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]);
@@@ -104,17 +104,52 @@ function validateSource(input, { maxItemsCap = 12000 } = {}) {
if (!e || typeof e !== 'object') throw new RegistryError(`endpoints[${i}] must be a URL or an object`);
const out = { url: httpUrl(e.url, `endpoints[${i}].url`) };
if (input.type === 'api') {
++ // A resolver walks several JSON endpoints of one API (HN Who is hiring), instead of reading one items_path:
++ // it has no field mapping of its own, so it does not go through the api mapping checks.
++ if (e.resolver !== undefined) {
++ for (const k of Object.keys(e)) {
++ if (!['url', 'format', 'resolver', 'item_kind', 'max_bytes', 'timeout_ms'].includes(k)) throw new RegistryError(`endpoints[${i}].${k} is not a resolver field`);
++ }
++ if (e.resolver !== 'hn-whoishiring') throw new RegistryError(`endpoints[${i}].resolver must be hn-whoishiring`);
++ if (e.format !== undefined && e.format !== 'json') throw new RegistryError(`endpoints[${i}].format must be json for a resolver`);
++ if (!/\.json(\?|$)/.test(out.url)) throw new RegistryError(`endpoints[${i}].resolver needs a .json endpoint`);
++ if (e.item_kind !== undefined && !/^[a-z][a-z0-9_]{1,39}$/.test(e.item_kind)) throw new RegistryError(`endpoints[${i}].item_kind is malformed`);
++ if (e.max_bytes !== undefined && (!Number.isInteger(e.max_bytes) || e.max_bytes < 1024 || e.max_bytes > 32 * 1024 * 1024)) throw new RegistryError(`endpoints[${i}].max_bytes must be an integer from 1024 to 33554432`);
++ if (e.timeout_ms !== undefined && (!Number.isInteger(e.timeout_ms) || e.timeout_ms < 1000 || e.timeout_ms > 120000)) throw new RegistryError(`endpoints[${i}].timeout_ms must be an integer from 1000 to 120000`);
++ Object.assign(out, { format: 'json', items_shape: 'array', items_path: '', item_kind: e.item_kind || 'job', fields: {}, extra: {}, resolver: e.resolver });
++ if (e.max_bytes !== undefined) out.max_bytes = e.max_bytes;
++ if (e.timeout_ms !== undefined) out.timeout_ms = e.timeout_ms;
++ return out;
++ }
const err = checkApiMapping(e);
if (err) throw new RegistryError(`endpoints[${i}]: ${err}`);
Object.assign(out, { format: e.format || 'json', items_shape: e.items_shape || 'array', items_path: e.items_path || '', item_kind: e.item_kind || 'record', fields: e.fields || {}, extra: e.extra || {} });
-- // `where` is only copied when given, so records without conditions stay byte-identical.
++ // `where` and its alias `filter` are only copied when given, so records without conditions stay byte-identical.
if (e.where) out.where = e.where;
++ if (e.filter) out.filter = e.filter;
// and `extra_lists` (tags, job types) the same way.
if (e.extra_lists) out.extra_lists = e.extra_lists;
if (e.duplicates !== undefined) out.duplicates = e.duplicates;
if (e.max_bytes !== undefined) out.max_bytes = e.max_bytes;
if (e.timeout_ms !== undefined) out.timeout_ms = e.timeout_ms;
if (e.request_content_type !== undefined) out.request_content_type = e.request_content_type;
++ } else if (input.type === 'bulk') {
++ if (endpoints.length !== 1) throw new RegistryError('a bulk source is one dataset file: exactly one endpoint');
++ const err = checkApiMapping(e, { bulk: true });
++ if (err) throw new RegistryError(`endpoints[${i}]: ${err}`);
++ for (const k of Object.keys(e)) {
++ if (!['url', 'format', 'item_kind', 'fields', 'extra', 'extra_lists', 'max_bytes', 'max_rows', 'timeout_ms'].includes(k)) throw new RegistryError(`endpoints[${i}].${k} is not a bulk field`);
++ }
++ Object.assign(out, {
++ format: e.format || 'csv',
++ item_kind: e.item_kind || 'record',
++ fields: e.fields || {},
++ extra: e.extra || {},
++ max_rows: e.max_rows === undefined ? 1000 : e.max_rows,
++ });
++ if (e.extra_lists) out.extra_lists = e.extra_lists;
++ if (e.max_bytes !== undefined) out.max_bytes = e.max_bytes;
++ if (e.timeout_ms !== undefined) out.timeout_ms = e.timeout_ms;
} else {
// A feed may read structured facts out of its items' prose (adapters/extract.js).
for (const k of Object.keys(e)) if (k !== 'url' && !(k === 'extract' && (input.type === 'rss' || input.type === 'atom'))) throw new RegistryError(`endpoints[${i}].${k} is only for api sources`);
@@@ -175,6 +210,8 @@ function validateSource(input, { maxItemsCap = 12000 } = {}) {
// 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');
++ // A bulk run reads up to max_rows and stops; it never sees the rest of the file, so "not seen" says nothing.
++ if (input.type === 'bulk' && record.retire_after_sec != null) throw new RegistryError('a bulk source reads a row limit per run; retire_after_sec would retire the rows it did not reach');
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) {
ddiff --git a/test/api-join.test.js b/test/api-join.test.js
iindex 5add77d..a1b998d 100644
---- a/test/api-join.test.js
++++ b/test/api-join.test.js
@@@ -163,7 +163,7 @@ t('a where pattern keeps text it matches; a number or a non-matching code is fil
t('checkMapping accepts selectors, patterns and max_bytes, and refuses their malformed variants', () => {
assert.strictEqual(api.checkMapping(selectEndpoint), null);
-- assert.strictEqual(api.checkMapping({ ...selectEndpoint, max_bytes: 32 * 1024 * 1024 }), null);
++ assert.strictEqual(api.checkMapping({ ...selectEndpoint, max_bytes: 64 * 1024 * 1024 }), null);
const extra = (x) => api.checkMapping({ fields: { identity: 'id' }, extra: { x } });
assert.strictEqual(extra('a[b=Monthly fees].c'), null, 'literal values may hold spaces');
assert.match(extra('a[b=1,c=2,d=3,e=4,f=5].x'), /malformed/, 'at most four conditions');
@@@ -176,7 +176,7 @@ t('checkMapping accepts selectors, patterns and max_bytes, and refuses their mal
assert.match(where({ pattern: 'x'.repeat(201) }), /at most 200/);
assert.match(where({ pattern: 'x', flags: 'i' }), /a pattern is/);
assert.match(where({ y: 1 }), /strings, numbers or booleans/, 'an object without pattern is still refused as before');
-- for (const max of [0, 1023, 32 * 1024 * 1024 + 1, 1.5, '5000000']) {
++ for (const max of [0, 1023, 64 * 1024 * 1024 + 1, 1.5, '5000000']) {
assert.match(api.checkMapping({ fields: { identity: 'id' }, max_bytes: max }), /max_bytes is an integer/, String(max));
}
});
ddiff --git a/test/api-tsv.test.js b/test/api-tsv.test.js
iindex ed09d24..d8ed657 100644
---- a/test/api-tsv.test.js
++++ b/test/api-tsv.test.js
@@@ -50,6 +50,29 @@ t('duplicates default to the first; maxItems counts distinct identities', () =>
assert.deepStrictEqual(capped.items.map((i) => [i.identity, i.fields.period]), [['APU0000708111', 'M09'], ['APU0000709112', 'M09']], 'a repeat of a kept series still replaces it at the cap');
});
++t('filter: the BLS CPI rent seed keeps one series\' rows, one item per month, by record position', () => {
++ const rent = {
++ format: 'tsv', items_path: '', item_kind: 'rent_cpi',
++ filter: { series_id: 'APU0000708111' },
++ fields: { identity: '$index', title: 'series_id' },
++ extra: { value: 'value', year: 'year', period: 'period' },
++ };
++ assert.strictEqual(api.checkMapping(rent), null);
++ const r = api.parse(BLS, { url: 'https://download.bls.gov/x', endpoint: rent });
++ assert.deepStrictEqual(r.items.map((i) => [i.identity, i.fields.value, i.fields.period]), [
++ ['0', '2.301', 'M07'], ['1', '2.272', 'M08'], ['2', '2.272', 'M09'],
++ ], 'only the series asked for; $index is the row\'s position in the whole file');
++ assert.strictEqual(r.meta.filtered, 4, 'the other series rows are filtered');
++ assert.strictEqual(r.skipped, 1, 'the malformed line');
++ // A filter and a where may both be given: every condition applies.
++ const both = api.parse(BLS, { url: 'https://download.bls.gov/x', endpoint: { ...rent, where: { year: '2026' } } });
++ assert.strictEqual(both.items.length, 3);
++ const none = api.parse(BLS, { url: 'https://download.bls.gov/x', endpoint: { ...rent, filter: { series_id: 'APU0000709999' } } });
++ assert.deepStrictEqual([none.items.length, none.meta.filtered], [0, 7]);
++ assert.match(api.checkMapping({ ...rent, filter: 'series_id' }), /filter must be an object/);
++ assert.match(api.checkMapping({ ...rent, filter: { 'bad path!': 'x' } }), /filter\.bad path! path is malformed/);
++});
++
t('mapping and body errors', () => {
assert.match(api.checkMapping({ ...blsEndpoint, format: 'csv' }), /format must be json, xml or tsv/);
assert.match(api.checkMapping({ ...blsEndpoint, items_path: 'rows' }), /items_path is ""/);
ddiff --git a/test/bulk.test.js b/test/bulk.test.js
nnew file mode 100644
iindex 0000000..f7d30db
---- /dev/null
++++ b/test/bulk.test.js
@@@ -0,0 +1,152 @@
++'use strict';
++/**
++ * The bulk dataset adapter and its ingest path: one streamed file, parsed as CSV, JSONL or TSV, with a row limit
++ * the stream stops at and a byte cap that refuses an oversize body. A large fixture is never read whole: the
++ * parser's write() returns false at the limit and the fetcher destroys the response.
++ */
++const assert = require('assert');
++const zlib = require('zlib');
++const { boot, site, sourceDef, suite } = require('./helpers');
++const { createParser, PARSER_VERSION } = require('../server/adapters/bulk');
++const { validateSource } = require('../server/registry');
++
++const t = suite('bulk');
++
++const csvEndpoint = {
++ format: 'csv', item_kind: 'product', max_rows: 10,
++ fields: { identity: 'code', title: 'product_name' },
++ extra: { energy_kcal: 'energy-kcal_100g', brands: 'brands' },
++};
++
++t('CSV: quoted commas and newlines survive a chunk boundary', () => {
++ const body = 'code,product_name,brands,energy-kcal_100g\r\n'
++ + '001,"Pate, noir",Bovetti,250\r\n'
++ + '002,"Two\nlines",Acme,120\r\n'
++ + '003,Plain,,90\r\n';
++ const parser = createParser({ endpoint: csvEndpoint, url: 'https://static.example.org/d.csv', maxItems: 100 });
++ for (const ch of Buffer.from(body)) parser.write(Buffer.from([ch])); // one byte at a time, quotes split too
++ const out = parser.end();
++ assert.strictEqual(PARSER_VERSION, 'bulk@1');
++ assert.deepStrictEqual(out.items.map((i) => [i.identity, i.title, i.fields.energy_kcal]), [
++ ['001', 'Pate, noir', '250'], ['002', 'Two lines', '120'], ['003', 'Plain', '90'],
++ ]);
++ assert.strictEqual(out.meta.format, 'csv');
++ assert.strictEqual(out.meta.rows, 3);
++});
++
++t('the row limit stops a large fixture without reading it whole', () => {
++ const endpoint = { ...csvEndpoint, max_rows: 1000 };
++ const parser = createParser({ endpoint, url: 'https://static.example.org/big.csv', maxItems: 5000 });
++ let fed = 0;
++ let wrote = 0;
++ const chunk = (from) => {
++ const lines = [];
++ if (from === 0) lines.push('code,product_name,brands,energy-kcal_100g');
++ for (let i = from; i < from + 100; i++) lines.push(`${i},Product ${i},Brand,${i % 900}`);
++ return Buffer.from(lines.join('\n') + '\n');
++ };
++ let from = 0;
++ for (;;) {
++ const buf = chunk(from);
++ from += 100;
++ fed += buf.length;
++ wrote++;
++ if (parser.write(buf) === false) break;
++ if (wrote > 100) throw new Error('the parser never stopped');
++ }
++ const out = parser.end();
++ assert.strictEqual(out.items.length, 1000, 'exactly the row limit');
++ assert.strictEqual(out.meta.capped, true);
++ assert.ok(fed < 200_000, `read ${fed} bytes, far less than a full multi-GB file`);
++ assert.ok(parser.full);
++});
++
++t('a byte cap is a recorded failure, and no item is made from a refused body', async () => {
++ const svc = await boot();
++ const big = 'code,product_name\n' + Array.from({ length: 2000 }, (_, i) => `${i},P${i}`).join('\n') + '\n';
++ const web = await site({
++ '/robots.txt': () => ({ body: 'User-agent: *\nAllow: /\n' }),
++ '/products.csv': (req, res) => {
++ res.writeHead(200, { 'Content-Type': 'text/csv' });
++ for (let i = 0; i < big.length; i += 1024) res.write(big.slice(i, i + 1024));
++ res.end();
++ },
++ });
++ try {
++ await svc.registry.create(sourceDef({
++ key: 'bulk-cap', type: 'bulk', category: 'food', enabled: true, max_items: 5000,
++ endpoints: [{ ...csvEndpoint, url: `${web.origin}/products.csv`, max_rows: 5000, max_bytes: 2048 }],
++ }), 'test');
++ const out = await svc.ingest.run('bulk-cap');
++ assert.strictEqual(out.runs[0].state, 'http_error');
++ assert.strictEqual(out.runs[0].error_code, 'too_large');
++ assert.strictEqual((await svc.db.prepare('SELECT COUNT(*) AS n FROM items WHERE source_key = ?').get('bulk-cap')).n, 0);
++ } finally { await svc.stop(); await web.close(); }
++});
++
++t('a gzipped dataset is decompressed while it streams; the row limit ends the download early', async () => {
++ const svc = await boot();
++ const rows = Array.from({ length: 5000 }, (_, i) => `${i},Product ${i},Brand,${i % 900}`);
++ const csv = 'code,product_name,brands,energy-kcal_100g\n' + rows.join('\n') + '\n';
++ const gz = zlib.gzipSync(csv);
++ let sent = 0;
++ const web = await site({
++ '/robots.txt': () => ({ body: 'User-agent: *\nAllow: /\n' }),
++ '/products.csv.gz': (req, res) => {
++ res.writeHead(200, { 'Content-Type': 'application/gzip' });
++ // Write it in pieces, and stop when the client has enough (the row limit destroys the response).
++ for (let i = 0; i < gz.length; i += 512) {
++ if (res.destroyed || res.writableEnded) break;
++ res.write(gz.slice(i, i + 512));
++ sent += Math.min(512, gz.length - i);
++ }
++ if (!res.destroyed && !res.writableEnded) res.end();
++ },
++ });
++ try {
++ await svc.registry.create(sourceDef({
++ key: 'bulk-gz', type: 'bulk', category: 'food', enabled: true,
++ endpoints: [{ ...csvEndpoint, url: `${web.origin}/products.csv.gz`, max_rows: 50, max_bytes: 8 * 1024 * 1024 }],
++ }), 'test');
++ const out = await svc.ingest.run('bulk-gz');
++ assert.deepStrictEqual([out.runs[0].state, out.runs[0].items_seen], ['ok', 50]);
++ assert.strictEqual((await svc.db.prepare('SELECT COUNT(*) AS n FROM items WHERE source_key = ?').get('bulk-gz')).n, 50);
++ assert.ok(sent <= gz.length, 'the server sent only what was asked for');
++ } finally { await svc.stop(); await web.close(); }
++});
++
++t('JSONL and TSV read nested paths and headers', () => {
++ const jsonl = '{"code":"a","name":"A","nutriments":{"energy-kcal":10}}\n{"code":"b","name":"B","nutriments":{"energy-kcal":20}}\nnot json\n';
++ const j = createParser({
++ endpoint: { format: 'jsonl', fields: { identity: 'code', title: 'name' }, extra: { energy_kcal: 'nutriments.energy-kcal' } },
++ url: 'https://static.example.org/d.jsonl', maxItems: 100,
++ });
++ j.write(Buffer.from(jsonl.slice(0, 37)));
++ j.write(Buffer.from(jsonl.slice(37)));
++ const jout = j.end();
++ assert.deepStrictEqual(jout.items.map((i) => [i.identity, i.fields.energy_kcal]), [['a', 10], ['b', 20]]);
++ assert.strictEqual(jout.skipped, 1, 'the malformed line is counted, not invented');
++
++ const tsv = 'series_id\tyear\tvalue\nCUUR0000SEHA\t2026\t450.402\nCUUR0000SEHA\t2025\t430.1\n';
++ const p = createParser({
++ endpoint: { format: 'tsv', fields: { identity: '$index', title: 'series_id' }, extra: { year: 'year', value: 'value' } },
++ url: 'https://download.bls.gov/x', maxItems: 100,
++ });
++ p.write(Buffer.from(tsv));
++ const tout = p.end();
++ assert.deepStrictEqual(tout.items.map((i) => [i.identity, i.fields.value]), [['0', '450.402'], ['1', '430.1']]);});
++
++t('the registry takes one bulk endpoint with a format, a row limit and a byte cap; it refuses a second file or retirement', () => {
++ const seed = sourceDef({
++ key: 'bulk-seed', type: 'bulk', category: 'food', enabled: false,
++ endpoints: [{ url: 'https://static.example.org/d.csv', format: 'csv', max_rows: 20, max_bytes: 4096, fields: { identity: 'code' } }],
++ });
++ const rec = validateSource(seed);
++ assert.deepStrictEqual([rec.type, rec.endpoints[0].format, rec.endpoints[0].max_rows], ['bulk', 'csv', 20]);
++ assert.throws(() => validateSource({ ...seed, endpoints: [...seed.endpoints, { ...seed.endpoints[0], url: 'https://static.example.org/e.csv' }] }), /exactly one endpoint/);
++ assert.throws(() => validateSource({ ...seed, retire_after_sec: 86400 }), /retire_after_sec/);
++ assert.throws(() => validateSource({ ...seed, endpoints: [{ ...seed.endpoints[0], where: { x: 1 } }] }), /not a bulk field/);
++ assert.throws(() => validateSource({ ...seed, endpoints: [{ ...seed.endpoints[0], format: 'json' }] }), /format must be csv, jsonl or tsv/);
++});
++
++t.run();
ddiff --git a/test/fixtures/hn/item-201.json b/test/fixtures/hn/item-201.json
nnew file mode 100644
iindex 0000000..6f1474b
---- /dev/null
++++ b/test/fixtures/hn/item-201.json
@@@ -0,0 +1,9 @@
++{
++ "id": 201,
++ "type": "comment",
++ "by": "acme_founder",
++ "parent": 101,
++ "time": 1793603600,
++ "text": "Acme Corp | Senior Backend Engineer | Remote (US)<p>We are hiring. Email [email protected] or call (415) 555-0123 to apply. <p>More at https://acme.example/jobs",
++ "kids": [301]
++}
ddiff --git a/test/fixtures/hn/item-202.json b/test/fixtures/hn/item-202.json
nnew file mode 100644
iindex 0000000..bb81b8b
---- /dev/null
++++ b/test/fixtures/hn/item-202.json
@@@ -0,0 +1,9 @@
++{
++ "id": 202,
++ "type": "comment",
++ "by": "gone",
++ "parent": 101,
++ "time": 1793603700,
++ "deleted": true,
++ "text": ""
++}
ddiff --git a/test/fixtures/hn/item-203.json b/test/fixtures/hn/item-203.json
nnew file mode 100644
iindex 0000000..dc2c489
---- /dev/null
++++ b/test/fixtures/hn/item-203.json
@@@ -0,0 +1,9 @@
++{
++ "id": 203,
++ "type": "comment",
++ "by": "beta_team",
++ "parent": 101,
++ "time": 1793603800,
++ "text": "Beta &amp; Co is hiring a product designer.<p>Contact +1 415-555-0199 or [email protected]. <p>Apply at https://beta.example/jobs",
++ "kids": []
++}
ddiff --git a/test/fixtures/hn/item-204.json b/test/fixtures/hn/item-204.json
nnew file mode 100644
iindex 0000000..c1de195
---- /dev/null
++++ b/test/fixtures/hn/item-204.json
@@@ -0,0 +1,9 @@
++{
++ "id": 204,
++ "type": "comment",
++ "by": "dead",
++ "parent": 101,
++ "time": 1793603900,
++ "dead": true,
++ "text": "Spam"
++}
ddiff --git a/test/fixtures/hn/item-301.json b/test/fixtures/hn/item-301.json
nnew file mode 100644
iindex 0000000..c572384
---- /dev/null
++++ b/test/fixtures/hn/item-301.json
@@@ -0,0 +1,9 @@
++{
++ "id": 301,
++ "type": "comment",
++ "by": "reply",
++ "parent": 201,
++ "time": 1793604000,
++ "text": "A reply that is not a job post: [email protected]",
++ "kids": []
++}
ddiff --git a/test/fixtures/hn/story-100.json b/test/fixtures/hn/story-100.json
nnew file mode 100644
iindex 0000000..85d5455
---- /dev/null
++++ b/test/fixtures/hn/story-100.json
@@@ -0,0 +1,10 @@
++{
++ "id": 100,
++ "type": "story",
++ "by": "whoishiring",
++ "time": 1791000000,
++ "title": "Ask HN: Who is hiring? (September 2026)",
++ "text": "",
++ "kids": [190],
++ "score": 280
++}
ddiff --git a/test/fixtures/hn/story-101.json b/test/fixtures/hn/story-101.json
nnew file mode 100644
iindex 0000000..be22d8d
---- /dev/null
++++ b/test/fixtures/hn/story-101.json
@@@ -0,0 +1,10 @@
++{
++ "id": 101,
++ "type": "story",
++ "by": "whoishiring",
++ "time": 1793600000,
++ "title": "Ask HN: Who is hiring? (October 2026)",
++ "text": "",
++ "kids": [201, 202, 203, 204],
++ "score": 312
++}
ddiff --git a/test/fixtures/hn/user-whoishiring.json b/test/fixtures/hn/user-whoishiring.json
nnew file mode 100644
iindex 0000000..06ace17
---- /dev/null
++++ b/test/fixtures/hn/user-whoishiring.json
@@@ -0,0 +1,5 @@
++{
++ "id": "whoishiring",
++ "created": 1300000000,
++ "submitted": [101, 100]
++}
ddiff --git a/test/hn.test.js b/test/hn.test.js
nnew file mode 100644
iindex 0000000..6c43bce
---- /dev/null
++++ b/test/hn.test.js
@@@ -0,0 +1,140 @@
++'use strict';
++/**
++ * The Hacker News "Who is hiring" resolver against recorded fixtures (no live call):
++ * - user/whoishiring.json → the newest matching story → its top-level kids only, never a reply's kids;
++ * - one job item per comment, the comment linked, the full text never stored;
++ * - the excerpt and title have emails and phones replaced;
++ * - every requested URL is a .json path, the API's robots rule.
++ */
++const assert = require('assert');
++const { resolve, stripPrivate, htmlToText, assertJsonUrl } = require('../server/adapters/hn');
++const { boot, fixture, site, sourceDef, suite } = require('./helpers');
++
++const t = suite('hn-whoishiring');
++const FILE = {
++ 'user/whoishiring.json': 'hn/user-whoishiring.json',
++ 'item/101.json': 'hn/story-101.json',
++ 'item/100.json': 'hn/story-100.json',
++ 'item/201.json': 'hn/item-201.json',
++ 'item/202.json': 'hn/item-202.json',
++ 'item/203.json': 'hn/item-203.json',
++ 'item/204.json': 'hn/item-204.json',
++ 'item/301.json': 'hn/item-301.json',
++};
++
++function recorded(extra = {}) {
++ const asked = [];
++ const getJson = async (url) => {
++ asked.push(url);
++ const key = String(url).replace('https://hacker-news.firebaseio.com/v0/', '');
++ if (extra[key]) return extra[key];
++ const name = FILE[key];
++ if (!name) throw new Error(`the resolver asked for a fixture that does not exist: ${url}`);
++ return JSON.parse(fixture(name));
++ };
++ return { asked, getJson };
++}
++
++const endpoint = { url: 'https://hacker-news.firebaseio.com/v0/user/whoishiring.json', max_bytes: 1048576 };
++
++t('walks user → newest story → top-level kids; one job per comment, no reply tree', async () => {
++ const { asked, getJson } = recorded();
++ const out = await resolve({ endpoint, getJson, maxItems: 100, summaryMax: 1000 });
++ assert.deepStrictEqual(out.items.map((i) => [i.identity, i.kind, i.canonical_url]), [
++ ['hn-201', 'job', 'https://news.ycombinator.com/item?id=201'],
++ ['hn-203', 'job', 'https://news.ycombinator.com/item?id=203'],
++ ], 'the deleted and dead comments are not job posts');
++ assert.strictEqual(out.skipped, 2);
++ assert.deepStrictEqual(asked, [
++ 'https://hacker-news.firebaseio.com/v0/user/whoishiring.json',
++ 'https://hacker-news.firebaseio.com/v0/item/101.json',
++ 'https://hacker-news.firebaseio.com/v0/item/201.json',
++ 'https://hacker-news.firebaseio.com/v0/item/202.json',
++ 'https://hacker-news.firebaseio.com/v0/item/203.json',
++ 'https://hacker-news.firebaseio.com/v0/item/204.json',
++ ], 'the newest story is the October one; reply 301 and older story 100 are never fetched');
++ assert.ok(asked.every((u) => new URL(u).pathname.endsWith('.json')), 'every request is a .json path');
++ const a = out.items[0];
++ assert.strictEqual(a.title, 'Acme Corp | Senior Backend Engineer | Remote (US)');
++ assert.deepStrictEqual(a.authors, ['acme_founder']);
++ assert.strictEqual(a.published_at, new Date(1793603600 * 1000).toISOString());
++ assert.deepStrictEqual(a.fields, {
++ story_id: '101',
++ story_title: 'Ask HN: Who is hiring? (October 2026)',
++ story_url: 'https://news.ycombinator.com/item?id=101',
++ });
++ assert.ok(!('text' in a) && !('html' in a), 'the full comment is not stored');
++});
++
++t('privacy: the excerpt and title carry [email] and [phone], never the address itself', async () => {
++ const { getJson } = recorded();
++ const out = await resolve({ endpoint, getJson, maxItems: 100, summaryMax: 1000 });
++ const [acme, beta] = out.items;
++ assert.ok(acme.summary.includes('[email]') && acme.summary.includes('[phone]'), acme.summary);
++ assert.ok(!acme.summary.includes('[email protected]') && !acme.summary.includes('555-0123'), acme.summary);
++ assert.ok(beta.summary.includes('[email]') && beta.summary.includes('[phone]'), beta.summary);
++ assert.ok(!beta.summary.includes('[email protected]') && !beta.summary.includes('555-0199'), beta.summary);
++ assert.ok(beta.title.includes('Beta & Co'), 'the HTML entity is text again');
++ assert.ok(!JSON.stringify(out.items).includes('@acme.example'));
++ assert.strictEqual(stripPrivate('mail me at [email protected] or call +44 20 7946 0958'), 'mail me at [email] or call [phone]');
++ assert.strictEqual(htmlToText('one<p>two &amp; three<br>four &#x27;five&#x27;'), "one\ntwo & three\nfour 'five'");
++});
++
++t('the .json rule and a thread with no Who is hiring story are refused', async () => {
++ assert.throws(() => assertJsonUrl('https://hacker-news.firebaseio.com/v0/item/1.html'), /only \.json/);
++ const { getJson } = recorded({
++ 'user/whoishiring.json': { id: 'whoishiring', submitted: [999] },
++ 'item/999.json': { id: 999, type: 'story', title: 'Ask HN: Who wants to be hired? (October 2026)', kids: [] },
++ });
++ await assert.rejects(
++ () => resolve({ endpoint, getJson, maxItems: 10, summaryMax: 1000 }),
++ (err) => err.code === 'parse_error' && /no "Ask HN: Who is hiring\?" story/.test(err.message),
++ );
++});
++
++t('the row limit stops at maxItems without fetching the rest', async () => {
++ const { asked, getJson } = recorded();
++ const out = await resolve({ endpoint, getJson, maxItems: 1, summaryMax: 1000 });
++ assert.strictEqual(out.items.length, 1);
++ assert.deepStrictEqual(asked.filter((u) => u.includes('/item/')), [
++ 'https://hacker-news.firebaseio.com/v0/item/101.json',
++ 'https://hacker-news.firebaseio.com/v0/item/201.json',
++ ]);
++});
++
++t('ingest: the resolver branch stores the jobs, links the comments and records its own parser version', async () => {
++ const svc = await boot();
++ const byPath = {
++ '/v0/user/whoishiring.json': 'hn/user-whoishiring.json',
++ '/v0/item/101.json': 'hn/story-101.json',
++ '/v0/item/201.json': 'hn/item-201.json',
++ '/v0/item/202.json': 'hn/item-202.json',
++ '/v0/item/203.json': 'hn/item-203.json',
++ '/v0/item/204.json': 'hn/item-204.json',
++ };
++ const web = await site({
++ '/robots.txt': () => ({ body: 'User-agent: *\nAllow: /*.json$\nAllow: /*.json?*$\nDisallow: /\n' }),
++ '*': (req, res, ctx) => {
++ const name = byPath[ctx.url.pathname];
++ if (!name) return { status: 404, body: 'no fixture' };
++ return { headers: { 'Content-Type': 'application/json' }, body: fixture(name) };
++ },
++ });
++ try {
++ await svc.registry.create(sourceDef({
++ key: 'hn-ingest', type: 'api', category: 'jobs', enabled: true, max_items: 100,
++ endpoints: [{ url: `${web.origin}/v0/user/whoishiring.json`, format: 'json', resolver: 'hn-whoishiring', item_kind: 'job', max_bytes: 1048576, timeout_ms: 10000 }],
++ }), 'test');
++ const out = await svc.ingest.run('hn-ingest');
++ assert.deepStrictEqual([out.runs[0].state, out.runs[0].parser_version, out.runs[0].items_seen], ['ok', 'hn-whoishiring@1', 2]);
++ const rows = await svc.db.prepare('SELECT identity, kind, canonical_url, summary, parser_version FROM items WHERE source_key = ? ORDER BY identity').all('hn-ingest');
++ assert.deepStrictEqual(rows.map((r) => [r.identity, r.kind, r.canonical_url]), [
++ ['hn-201', 'job', 'https://news.ycombinator.com/item?id=201'],
++ ['hn-203', 'job', 'https://news.ycombinator.com/item?id=203'],
++ ]);
++ assert.ok(rows.every((r) => r.parser_version === 'hn-whoishiring@1'));
++ assert.ok(!rows.some((r) => /jobs@acme\.example|555-0123|beta@example\.org/.test(r.summary)), 'no email or phone reached the database');
++ } finally { await svc.stop(); await web.close(); }
++});
++
++t.run();