<!-- nornir:class: design -->
# korp-collectors — the open-data collector leaf
> **What it is.** The standalone LEAF of korp's ingest layer: one object-safe
> `Collector` trait + per-access-method collectors that parse open data (pure-Rust →
> Arrow), land it via an Iceberg `BatchSink` (skade), and declare a typed
> `source_target()` so the test-matrix can iterate `(collector, source, target)`.
> Depends on **neither korp nor facett** — both consume it; it breaks the facett⇄korp cycle.
## Function axis
- **f0** — the `Collector` trait (object-safe) + `registry()` of all collectors.
- **f1** — per-method sources: `dcat-ap-se · pxweb/jsonstat2 · eurostat-sdmx (SDMX-JSON 2.1) ·
entsoe (price/load) ·
entsoe_generation (A75 per-type) · kolada · smhi-metobs · smhi-weather (stations) ·
polisen · trafikverket-trainpos/trains/roads · osm (RegionSource) · osm-overpass (POI) ·
ais (boats) · marinetraffic-ais (AISStream) · opensky-flights + opensky-europe-bbox
(planes) · gbfs-citybikes · bolagsverket-organisationer (company registry) ·
nordpool (electricity spot prices) · energy-charts-power (generation-mix parallel arrays) ·
usgs-earthquakes (GeoJSON seismicity) ·
wfs (OGC GML features) · rss-atom (RSS/Atom syndication feeds) ·
openaq (air-quality measurements) ·
gtfs-stops/routes/trips (transit feed zip) · gtfs-rt-vehicles (GTFS-Realtime protobuf
vehicle positions) · custom REST/JSON`.
- **f2** — parse: `parse_records` / `parse_records_bounded` → Arrow `RecordBatch` (pure Rust).
- **f3** — sink: `BatchSink` → **skade** Iceberg table (feature `skade-sink`); geo collectors
also target a **facett** map scene (e.g. `osm3d-flight` for opensky).
- **f4** — coverage: every collector emits a `descriptor_declared` **testmatrix** cell
(`emit_registry_descriptors()`), and each has an inject/assert test + an offline-safe
real-or-synthetic sample. Green ⟺ every declared source is exercised in code.
## Collectors (Direction::Source)
| Collector | Source | Lands |
|---|---|---|
| entsoe | ENTSO-E transparency (price/load) | warehouse table |
| **entsoe_generation** | ENTSO-E `documentType=A75` (generation per type, psrType) | warehouse table |
| ais | AIS positions (boats) | position table → `osm3d-flight` scene |
| **marinetraffic-ais** | AISStream.io nested `PositionReport` feed (boats) | ship position table → `osm3d-flight` scene |
| opensky-flights | OpenSky `states/all` (planes) | flight position table → `osm3d-flight` scene |
| **opensky-europe-bbox** | OpenSky `states/all` bbox-scoped to Europe | flight position table → `osm3d-flight` scene |
| trafikverket-trainpos / **trafikverket_trains** | Trafikverket `TrainPosition` XML query (trains) | position table → `osm3d-flight` scene |
| **trafikverket-roads** | Trafikverket NVDB `RoadGeometry` (WKT LINESTRING) | ways table → `osm3d` scene |
| **gbfs-citybikes** | GBFS `free_bike_status` (dockless bikes) | position table → `osm3d-flight` scene |
| **osm-overpass** | OSM Overpass QL nodes (selectable POI query) | point feature table → `osm3d-poi` scene |
| **smhi-weather** | SMHI metobs station-set (station geometry) | point table → `osm3d` scene |
| osm | osm-katana RegionSource | ways → `osm3d` map layer |
| **bolagsverket-organisationer** | Bolagsverket Näringslivsregistret (orgnr spine) | company registry warehouse table |
| **nordpool** | Nord Pool day-ahead spot prices (`multiAreaEntries` period×zone) | price table per bidding zone (SE1–SE4/NO/DK/FI) |
| **energy-charts-power** | Energy-Charts (Fraunhofer ISE) public-power generation mix **JSON** — parallel `unix_seconds[]` axis ZIPPED against each `production_types[].data[]` by index (public CC-BY; Europe-wide via `?country=`) | generation-mix table (time, production_type, power_mw) warehouse table |
| **usgs-earthquakes** | USGS real-time earthquake **GeoJSON** `FeatureCollection` (public/no-auth; each Point feature) | quake position table (id, place, magnitude, depth, lon/lat, epoch-ms time) → `osm3d` scene |
| **wfs** | OGC WFS `GetFeature` GML FeatureCollection (Lantmäteriet/SGU geodata) | feature table (WKT geometry + properties JSON) → `osm3d` scene |
| **rss-atom** | RSS 2.0 / Atom 1.0 syndication feed (Krisinformation/SMHI/agency alert + news streams; dialect-agnostic) | feed-item table (feed_title, id, title, link, summary, author, published, categories JSON) warehouse table |
| **openaq-air-quality** | OpenAQ air-quality measurements **JSON** `results[]` (public CC-BY; free API key via `X-API-Key`; nested `coordinates{lat,lon}` + `date.utc` per reading) | air-quality measurement table (location_id, location, parameter, value, unit, lon/lat, country, city, UTC time) → `osm3d` scene |
| **gtfs-stops / gtfs-routes / gtfs-trips** | GTFS static transit feed zip (pure-Rust ZIP + csv; feature `gtfs`) | stops (→ `osm3d` scene) / routes / trips warehouse tables |
| **gtfs-rt-vehicles** | GTFS-Realtime `VehiclePositions` **protobuf** `FeedMessage` (hand-rolled varint/length-delimited reader — no `prost` dep; a binary wire shape no other collector decodes) | live vehicle position table (entity/vehicle/route/trip id, lat/lon, bearing, speed, timestamp, stop) → `osm3d-flight` scene |
| **eurostat-sdmx** | Eurostat SDMX-JSON 2.1 (series/observation split, colon-key decode) | EU stats cube → per-`(geo × time × measure)` value rows warehouse table |
| dcat-ap-se · pxweb · kolada · smhi-metobs · polisen | SE open-data portals | warehouse tables |
## Laws (this leaf)
- Leaf: no `korp`/`facett` dep — ever.
- RAGNARÖK: a collect landing 0 rows is RED unless `CollectStats.reason` says why.
The verdict is **computable**, not a convention: `CollectStats::ragnarok()` →
`Rows | ExplainedEmpty | Red`, with `is_red()` / `ragnarok_check()` as the gate
and `run_strict()` as the one-call runner that returns `Err` on an unexplained
empty. `run()`'s own annotation uses the recognisable `RAGNAROK_UNEXPLAINED`
sentinel, so the runner can never launder a silent empty into an honest one.
- **The registry sweeps** (`tests/collectors_registry_sweep.rs`) assert the two
laws over EVERY collector in `registry()` rather than the handful with a
hand-written test:
1. `sample::for_collector(name)` resolves an offline-safe sample for every
registered name (RED when a collector lands without one);
2. every collector fed an EMPTY-but-shape-preserving source (its own sample with
the arrays emptied, so its records pointer still resolves) either errors or
returns 0 rows **with a reason** — never `Ok` with 0 rows and no reason;
3. every collector `collect()`s its own sample into ≥1 row — RED when a parse
breaks. One `functional_status` cell per `(collector, law)`.
- Real-or-fake samples: attempt a short-timeout live fetch, fall back to an embedded
`Synthetic` fixture — tests/benches never hang, never fail for want of network
(`KORP_COLLECTORS_OFFLINE=1` forces synthetic).
- The RECORDED sample URL is credential-free: some live sample endpoints carry a
token IN THE QUERY (ENTSO-E `securityToken`, Destatis GENESIS `token`), and
`SampleOrigin::Real{url}` is rendered by `label()` into durable test-matrix rows,
so `sample::redact_url` scrubs known secret query params (and userinfo passwords)
at the single point the fetch URL becomes recorded identity.
- Auth resolves through the ASYNC seam: every collector calls
`cfg.auth.resolve_async().await?`, never the sync `resolve()`. The sync form
cannot serve `AuthConfig::OAuth2ClientCredentials` (the token is a network
round-trip) and errors out; `resolve_async` delegates every other variant back
to it, so using it always is free. Swept by `tests/collectors_auth.rs`.
- testmatrix cells emitted under `--features testmatrix` (OFF = no-op, no nornir dep).
## Open work (see `korp/.nornir/korp-worklist.md`)
- ✅ Europe-extent bbox (`lamin/lomin/lamax/lomax`) opensky variant landed (`opensky-europe-bbox`).
- The **UI proof**: an Ingest/Tools tab that runs a collector as a **jera job** — ships
ONE `nornir-robotui` scenario (headless + watch:0) + ONE testmatrix cell. Blocked on
extracting a `facett-korp` pane from korp's welded `main.rs` (single-source P0).