nordisk/korp-collectors
src/rest.rs
//! **#5 Generic REST/JSON adapter framework** — a thin, config-driven adapter
//! over the shared HTTP core that maps a JSON response → Arrow. One generic
//! [`RestCollector`] + a per-source field map covers the long tail of high-value
//! JSON APIs (SMHI, Polisen, Kolada, Riksbanken, JobTech, Riksdagen, …). Three
//! concrete adapters ship as proof: [`RestCollector::polisen`],
//! [`RestCollector::kolada`], [`RestCollector::smhi_metobs`].
//!
//! A field map is `(column, JSON-pointer-within-record, type)`; the records live
//! at `records_pointer` (RFC-6901, `""` = the whole doc is the array). Offline
//! contract: [`parse_records`] turns a canned JSON payload → Arrow, asserted as
//! DATA.
use anyhow::{Result, anyhow};
use arrow_array::{Array, ArrayRef, RecordBatch};
use arrow_schema::{DataType, Schema};
use serde_json::Value;
use crate::freshness::MarkOrder;
use super::arrow_util::{batch, f64_col, i64_col, schema, str_col};
use super::{
AuthConfig, AuthKind, BatchSink, Bounds, Checkpoint, CollectStats, Collector, CollectorCfg,
CollectorMeta, CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
// ─────────────────────────────────────────────────────────────────────────────
// Endpoints that more than one method has to agree on — ONE writer each, so a
// preset, its `source_target` and its default auth can never drift apart.
// ─────────────────────────────────────────────────────────────────────────────
/// Fintraffic Digitraffic marine AIS locations. Measured live 2026-09-03: HTTP 200,
/// 40 400 B gzipped / 417 567 B inflated, 1001 GeoJSON features.
pub const DIGITRAFFIC_AIS_URL: &str = "https://meri.digitraffic.fi/api/ais/v1/locations";
/// The value Digitraffic's mandatory `Digitraffic-User` identification header carries.
/// Fintraffic asks for a string that identifies the client, not a credential; without
/// the header the gateway answers **502** (measured 2026-09-03).
pub const DIGITRAFFIC_USER: &str = "korp-collectors/0.1";
/// Bolagsverket **Värdefulla datamängder** (free HVD API) — `POST /organisationer`.
/// The OpenAPI's own `servers[0]` plus the documented path.
pub const BOLAGSVERKET_ORGANISATIONER_URL: &str =
"https://gw.api.bolagsverket.se/vardefulla-datamangder/v1/organisationer";
/// Bolagsverket's OAuth2 **client-credentials token endpoint**.
///
/// NOT on the gateway host: the OpenAPI's relative `tokenUrl: "/oauth2/token"` resolves
/// to `gw.api.bolagsverket.se/oauth2/token`, measured **404** on 2026-09-03 (as are
/// `/token` and `/authorize` there). The portal host answers as a real OAuth2 endpoint
/// (401 `invalid_client`, `WWW-Authenticate: Basic realm=portal.api.bolagsverket.se`),
/// and it reads the client id/secret from the FORM BODY — which is what
/// [`crate::http::oauth2_client_credentials_token`] sends.
pub const BOLAGSVERKET_TOKEN_URL: &str = "https://portal.api.bolagsverket.se/oauth2/token";
/// The Bolagsverket client-id env var. Declared by name so a missing credential is a
/// refusal that says WHICH variable, never a 401 at the wire.
pub const BOLAGSVERKET_CLIENT_ID_ENV: &str = "BOLAGSVERKET_CLIENT_ID";
/// The Bolagsverket client-secret env var.
pub const BOLAGSVERKET_CLIENT_SECRET_ENV: &str = "BOLAGSVERKET_CLIENT_SECRET";
/// **Domstolsverket's rättspraxis publication list — the whole url, sort included.**
///
/// `GET /api/v1/publiceringar` on the PUH API. Keyless: measured live 2026-09-05, HTTP
/// 200, `x-total-count: 17340` publications. The OpenAPI that documents every parameter
/// below is [`DOMSTOL_OPENAPI_URL`]; none of these names were guessable, and ten plausible
/// guesses (`size`, `limit`, `antal`, `fran`, `fromDate`, `domstolKod`, …) were all
/// measured to be silently IGNORED before the spec was found.
///
/// **`sortorder=publiceringstid` is the load-bearing half of this url**, and it is not a
/// preference. The API's default sort is `avgorandedatum` — a DATE, ten characters, with
/// no tiebreak — so its paging is not a total order and a crawl straddles the ties.
/// Measured 2026-09-05 over the identical 610-publication slice
/// (`publicerad_fran_och_med=2026-01-01`, 7 pages of 100):
///
/// | sort | rows returned | distinct `id` | duplicated | silently MISSED |
/// |---|---|---|---|---|
/// | default (`avgorandedatum`) | 610 | **607** | 3 | **3** |
/// | `sortorder=publiceringstid` | 610 | **610** | 0 | 0 |
///
/// Every duplicate was the last record of one page reappearing as the first record of
/// the next, on three of the six boundaries — and the `publiceringstid` crawl returned
/// every id the default crawl found **plus the three it dropped**. 0.49 % of a legal
/// corpus vanishing with nothing red anywhere is precisely the failure mode that makes a
/// warehouse untrustworthy, and the fix costs one query parameter.
///
/// `publiceringstid` is a minute-granular timestamp, so it is a much finer order than a
/// date — but still not a proven total one, which is why the preset ALSO deduplicates on
/// `id` and reports what it dropped.
///
/// `pagesize=100` is the server's ceiling, not a choice: `pagesize=500` and
/// `pagesize=1000` return byte-identical 100-record responses (measured 2026-09-05).
pub const DOMSTOL_RATTSPRAXIS_URL: &str =
"https://rattspraxis.etjanst.domstol.se/api/v1/publiceringar\
?sortorder=publiceringstid&pagesize=100";
/// The page size [`DOMSTOL_RATTSPRAXIS_URL`] actually receives — the server's own
/// clamp. One writer, because [`PageWalk::page_size`] and the url must agree or the
/// walk stops on its first page.
pub const DOMSTOL_PAGE_SIZE: usize = 100;
/// Domstolsverket's own OpenAPI for the PUH (publicerings-) API, linked from
/// `domstol.se`'s öppna-data page. It is the only place the query parameters are named.
pub const DOMSTOL_OPENAPI_URL: &str =
"https://rattspraxis.etjanst.domstol.se/openapi/puh-openapi.yaml";
/// **The SCB Företagsregistret API key env var** — the credential lane EC O6 is blocked
/// on, named so the refusal can name it.
///
/// SCB's allmänna företagsregister is a REST/JSON service over the per-company register:
/// one row per organisationsnummer and per arbetsställe, with SNI code, address and an
/// employee-count band. **It is free** since 2025-06-26 and still credential-gated —
/// requested by mail to [`SCB_FORETAGSREGISTRET_CONTACT`], which is where the owner's
/// request sits.
///
/// SCB's own page states the authorisation is a CERTIFICATE plus password today and
/// "kommer att bytas från nuvarande certifikat till API-nyckel" in **September 2026** —
/// which is this month, and is why exactly one secret is declared rather than three. See
/// [`RestCollector::requirements`] for what happens if a certificate arrives instead.
pub const SCB_FORETAGSREGISTRET_KEY_ENV: &str = "SCB_FORETAGSREGISTRET_KEY";
/// Where the SCB Företagsregistret credential is requested. Published by SCB on the
/// *Avgiftsfria uppgifter i företagsregistret* page; quoted in the refusal so an
/// operator reading "cannot run" also reads what to do about it.
pub const SCB_FORETAGSREGISTRET_CONTACT: &str = "scbforetag@scb.se";
/// **The Företagsregistret base url is NOT PUBLISHED**, so this is an RFC 2606
/// placeholder and `Requirements::readiness` reports it as one.
///
/// This is not laziness and not a TODO. SCB publishes that the service is "en
/// REST-tjänst baserad på http och json eller XML" and issues the endpoint together with
/// the credentials; it appears on no public page. Inventing a plausible host would be
/// worse than useless — it would turn a refusal an operator can act on into a DNS error
/// they cannot. The honest declaration is "the feed URL must be configured", which is
/// exactly the state `rss-atom`, `wfs` and `gtfs-rt-vehicles` already live in.
pub const SCB_FORETAGSREGISTRET_URL: &str = "https://foretagsregistret.example/api";
/// Arrow type a JSON field maps to.
#[derive(Debug, Clone, Copy)]
pub enum FieldType {
Str,
I64,
F64,
/// **A JSON ARRAY of scalars, joined into one Utf8 cell** with `"; "`.
///
/// The honest shape for a repeated field a relational row has to carry once:
/// Domstolsverket's `malNummerLista` holds one to four case numbers (measured
/// 2026-09-05: 24 of 610 publications carry more than one, the widest four), and
/// [`FieldType::Str`] would land the serde debug spelling `["M 1-25","M 2-25"]` —
/// JSON text inside a text column, which every downstream query then has to parse
/// back out.
///
/// An EMPTY array lands NULL, not `""`: "this publication cites no statute" and
/// "this publication cites the empty statute" must not be the same value.
Joined,
/// **The CHARACTER COUNT of a string field, without landing the string.**
///
/// For a field that is too large to land but whose presence and size are real
/// information. Domstolsverket's `innehall` (the full judgment text) is the case:
/// measured 2026-09-05 over a 610-publication slice it is present on 258 of them,
/// median 18 944 characters — and ONE single judgment carries **24 552 235**, which
/// is 27.5 MB of the slice's 34.8 MB in one cell. Landing it would put roughly a
/// gigabyte in one `RecordBatch` on a first full crawl; landing its size says the
/// text exists and how big it is, and the document itself is reachable as the PDF
/// bilaga.
///
/// A missing / non-string field lands NULL (unknown), never `0` (known to be empty).
TextLen,
}
impl FieldType {
/// The Arrow `DataType` this field maps to.
pub fn data_type(self) -> DataType {
match self {
FieldType::Str | FieldType::Joined => DataType::Utf8,
FieldType::I64 | FieldType::TextLen => DataType::Int64,
FieldType::F64 => DataType::Float64,
}
}
}
/// The separator [`FieldType::Joined`] uses. `"; "` and not `","`, because the values
/// it joins (case numbers, keywords, legal areas) contain commas of their own.
pub const JOIN_SEP: &str = "; ";
/// Join a JSON array of scalars into one cell — see [`FieldType::Joined`].
fn as_joined(v: &Value) -> Option<String> {
match v {
Value::Array(a) => {
let parts: Vec<String> = a.iter().filter_map(as_string).collect();
(!parts.is_empty()).then(|| parts.join(JOIN_SEP))
}
// A scalar where the source usually serves an array is still a value, and
// dropping it would be a silent hole. One element joins to itself.
other => as_string(other),
}
}
/// The character count of a string field — see [`FieldType::TextLen`].
fn as_text_len(v: &Value) -> Option<i64> {
v.as_str().map(|s| s.chars().count() as i64)
}
/// One column ← `pointer` (relative to each record) as `ty`.
#[derive(Debug, Clone)]
pub struct FieldMap {
pub column: String,
pub pointer: String,
pub ty: FieldType,
}
impl FieldMap {
pub fn new(column: &str, pointer: &str, ty: FieldType) -> Self {
Self {
column: column.into(),
pointer: pointer.into(),
ty,
}
}
}
fn as_string(v: &Value) -> Option<String> {
match v {
Value::String(s) => Some(s.clone()),
Value::Null => None,
other => Some(other.to_string()),
}
}
fn as_i64(v: &Value) -> Option<i64> {
v.as_i64()
.or_else(|| v.as_str().and_then(|s| s.trim().parse().ok()))
}
fn as_f64(v: &Value) -> Option<f64> {
v.as_f64()
.or_else(|| v.as_str().and_then(|s| s.trim().parse().ok()))
}
/// Resolve the records array: `records_pointer == ""` means the document itself
/// is the array; otherwise it's the array at that RFC-6901 pointer.
fn records<'a>(json: &'a Value, ptr: &str) -> Result<&'a Vec<Value>> {
let node = if ptr.is_empty() {
json
} else {
json.pointer(ptr)
.ok_or_else(|| anyhow!("records pointer `{ptr}` not found"))?
};
node.as_array()
.ok_or_else(|| anyhow!("records at `{ptr}` is not an array"))
}
/// **The next-page url a response carries**, by RFC-6901 pointer.
///
/// `None` for an absent, null, non-string or empty value — all four mean "no more
/// pages", and a JSON:API list's last page really does serve `links: {}`. Trimmed,
/// because a url with trailing whitespace is a url nothing answers.
fn next_link(json: &Value, ptr: &str) -> Option<String> {
json.pointer(ptr)
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string)
}
/// **Join several run notes into one `CollectStats::reason`**, or `None` when there is
/// nothing to say. A run can have more than one thing worth reporting (it hit the page
/// ceiling AND it deduplicated), and a reason field that could hold only the first would
/// silently drop the second.
pub(crate) fn join_reasons(parts: &[Option<String>]) -> Option<String> {
let kept: Vec<&str> = parts.iter().filter_map(|p| p.as_deref()).collect();
(!kept.is_empty()).then(|| kept.join(" · "))
}
/// **Append the declared derived columns** to a batch the field map built.
///
/// A source column the batch does not carry is a loud `Err` — a derived JOIN KEY that
/// silently came out all-NULL is the failure this exists to prevent, not a smaller
/// version of it.
fn apply_derived(rb: RecordBatch, derived: &[Derive]) -> Result<RecordBatch> {
if derived.is_empty() {
return Ok(rb);
}
let mut fields: Vec<arrow_schema::Field> =
rb.schema().fields().iter().map(|f| (**f).clone()).collect();
let mut cols: Vec<ArrayRef> = rb.columns().to_vec();
for d in derived {
let src = rb
.column_by_name(&d.from)
.ok_or_else(|| anyhow!("derived column `{}` reads `{}`, which the field map does not produce", d.column, d.from))?;
let src = src
.as_any()
.downcast_ref::<arrow_array::StringArray>()
.ok_or_else(|| anyhow!("derived column `{}` needs `{}` to be Utf8", d.column, d.from))?;
let out: arrow_array::StringArray = (0..src.len())
.map(|i| {
if src.is_null(i) {
return None;
}
match d.how {
// A value with no digits at all yields NULL, not "": an
// all-punctuation source value would otherwise become the empty
// string and every such row would key alike.
DeriveHow::DigitsOnly => {
let v: String = src.value(i).chars().filter(char::is_ascii_digit).collect();
(!v.is_empty()).then_some(v)
}
}
})
.collect();
fields.push(arrow_schema::Field::new(&d.column, DataType::Utf8, true));
cols.push(std::sync::Arc::new(out) as ArrayRef);
}
Ok(RecordBatch::try_new(
std::sync::Arc::new(arrow_schema::Schema::new(fields)),
cols,
)?)
}
/// Build an Arrow batch from a slice of record refs via a field map.
fn build_batch(recs: &[&Value], fields: &[FieldMap]) -> Result<RecordBatch> {
let mut columns: Vec<ArrayRef> = Vec::with_capacity(fields.len());
for f in fields {
let cells = recs.iter().map(|r| {
// pointer "" / "/x" — empty means the record itself.
if f.pointer.is_empty() {
Some(*r)
} else {
r.pointer(&f.pointer)
}
});
let col: ArrayRef = match f.ty {
FieldType::Str => str_col(&cells.map(|c| c.and_then(as_string)).collect::<Vec<_>>()),
FieldType::I64 => i64_col(&cells.map(|c| c.and_then(as_i64)).collect::<Vec<_>>()),
FieldType::F64 => f64_col(&cells.map(|c| c.and_then(as_f64)).collect::<Vec<_>>()),
FieldType::Joined => {
str_col(&cells.map(|c| c.and_then(as_joined)).collect::<Vec<_>>())
}
FieldType::TextLen => {
i64_col(&cells.map(|c| c.and_then(as_text_len)).collect::<Vec<_>>())
}
};
columns.push(col);
}
let fdefs: Vec<(&str, DataType, bool)> = fields
.iter()
.map(|f| (f.column.as_str(), f.ty.data_type(), true))
.collect();
batch(&schema(&fdefs), columns)
}
/// Map a JSON payload → Arrow via a field map. The offline contract.
pub fn parse_records(
json: &Value,
records_pointer: &str,
fields: &[FieldMap],
) -> Result<RecordBatch> {
let recs = records(json, records_pointer)?;
let refs: Vec<&Value> = recs.iter().collect();
build_batch(&refs, fields)
}
/// Keep only records whose datetime at `ts_pointer` (RFC-6901, relative to each
/// record) falls within `bounds` — the incremental / CDC filter. Bounds are
/// **inclusive** and compared lexically (ISO-8601 timestamps sort by value). A
/// record whose timestamp is missing / non-string is DROPPED whenever any bound
/// is set (conservative: an unbounded record can't be proven in-range); with
/// both bounds `None` every record is kept (identity — the seam is a no-op).
pub fn filter_by_bounds<'a>(
recs: &'a [Value],
ts_pointer: &str,
bounds: &Bounds,
) -> Vec<&'a Value> {
filter_by_bounds_ordered(recs, ts_pointer, bounds, MarkOrder::Lexical)
}
/// [`filter_by_bounds`] with the comparison named.
///
/// The lexical default was silently wrong for the one source this crate reads most:
/// polisen stamps `2026-09-03 1:23:41 +02:00` with a **single-digit hour**, so
/// `"…9:58:00…" > "…11:23:41…"` as text and a lexical watermark drops every event after
/// 09:59 until the next midnight. [`MarkOrder`] is the single writer of "is this
/// newer" (see [`crate::freshness`]); this function and the file-series walk both go
/// through it so they cannot disagree.
///
/// The bound is **strictly after** `since` when an order other than `Lexical` is used,
/// because a watermark is a record the last run already landed: `>=` would re-land it
/// on every run and put the 500 → 1 504 defect straight back.
pub fn filter_by_bounds_ordered<'a>(
recs: &'a [Value],
ts_pointer: &str,
bounds: &Bounds,
order: MarkOrder,
) -> Vec<&'a Value> {
if bounds.since.is_none() && bounds.until.is_none() {
return recs.iter().collect();
}
let strict = order != MarkOrder::Lexical;
recs.iter()
.filter(|r| {
let ts = if ts_pointer.is_empty() {
Some(*r)
} else {
r.pointer(ts_pointer)
}
.and_then(|v| v.as_str());
match ts {
Some(ts) => {
let after = bounds.since.as_deref().is_none_or(|s| {
if strict {
order.is_after(ts, Some(s))
} else {
ts >= s
}
});
let before = bounds
.until
.as_deref()
.is_none_or(|u| order.sort_key(ts) <= order.sort_key(u));
after && before
}
None => false,
}
})
.collect()
}
/// The **newest** timestamp at `ts_pointer` across `recs` under `order` — the CDC
/// high-watermark a [`Collector::checkpoint`] reports as the resume mark.
fn max_mark(recs: &[&Value], ts_pointer: &str, order: MarkOrder) -> Option<String> {
let vals: Vec<&str> = recs
.iter()
.filter_map(|r| {
if ts_pointer.is_empty() {
Some(*r)
} else {
r.pointer(ts_pointer)
}
.and_then(|v| v.as_str())
})
.collect();
order.newest(vals)
}
/// Map a JSON payload → Arrow, first NARROWING the records to `bounds` by the
/// datetime at `ts_pointer` (the incremental collect path). The offline contract
/// for the CDC seam.
pub fn parse_records_bounded(
json: &Value,
records_pointer: &str,
fields: &[FieldMap],
ts_pointer: &str,
bounds: &Bounds,
) -> Result<RecordBatch> {
let recs = records(json, records_pointer)?;
let kept = filter_by_bounds(recs, ts_pointer, bounds);
build_batch(&kept, fields)
}
/// The config-driven REST/JSON collector.
pub struct RestCollector {
name: String,
description: String,
license: String,
gdpr: GdprClass,
auth: AuthKind,
join_keys: Vec<String>,
records_pointer: String,
fields: Vec<FieldMap>,
/// RFC-6901 pointer (relative to each record) to the datetime field the CDC
/// [`Bounds`] filter narrows by. `None` → the collector ignores bounds (no
/// declared incremental key). Set via [`RestCollector::with_bounds_field`].
ts_pointer: Option<String>,
/// The freshness declaration, when this adapter knows how its source changes.
/// `None` falls back to the derived default over
/// [`ts_pointer`](RestCollector::bounds_field) — see
/// [`Collector::freshness`](crate::Collector::freshness).
freshness: Option<crate::freshness::Freshness>,
/// The business key, when the adapter knows a narrower identity than its declared
/// join keys. `None` falls back to the derived default.
business_key: Option<crate::history::BusinessKey>,
/// **RFC-6901 pointer to the NEXT-PAGE url**, when this source paginates by
/// cursor. `None` ⇒ one request per run, which is right for every source whose
/// whole answer fits in one response.
///
/// GLEIF serves **119 764** Swedish LEI records (measured live 2026-09-04) 200 at
/// a time behind a JSON:API `links.next`, and a collector that read one page would
/// land 200 and report success. For a JOIN KEY that is the worst available outcome:
/// every company beyond page one reads as a company with no LEI, and nothing is
/// red anywhere.
next_pointer: Option<String>,
/// **Page-NUMBER pagination**, for a source that serves no cursor at all. `None` ⇒
/// this preset either follows [`next_pointer`](RestCollector::next_pointer) or
/// reads one response. See [`PageWalk`].
page_walk: Option<PageWalk>,
/// **RFC-6901 pointer to the record's own identity**, when the crawl must drop a
/// record it has already read. `None` ⇒ every record of every page lands, which is
/// right for a source whose paging is a stable total order.
///
/// Domstolsverket's is not. Measured 2026-09-05: a 7-page crawl of its 610-record
/// 2026 slice under the DEFAULT sort returned 610 rows carrying only **607 distinct
/// `id`s** — the last record of a page reappeared as the first record of the next
/// on three of the six boundaries. See [`DOMSTOL_RATTSPRAXIS_URL`] for the sort that
/// fixes it and why this stays on anyway.
dedupe_pointer: Option<String>,
/// The page ceiling for a cursor crawl. A crawl has to stop somewhere and a source
/// that grew a cycle must not become an infinite loop; hitting it is REPORTED in
/// `CollectStats::reason` even though the run landed rows, so nobody reads a
/// truncated table as a complete one.
max_pages: usize,
/// Columns derived from another column rather than read from the record — the
/// normalisations a JOIN KEY needs. See [`Derive`].
derived: Vec<Derive>,
/// What the last [`Collector::collect`] read — the watermark, the validators, the
/// bytes it cost and whether the source proved unchanged. Reported back by
/// [`Collector::checkpoint`], and persisted by the RUNNER only after a
/// successful land.
last_read: std::sync::Mutex<Option<Checkpoint>>,
}
/// **A column computed from another column** — the normalisation a join key needs.
///
/// GLEIF serves an orgnr as `559586-2557`; Bolagsverket's `orgnr` column carries no
/// hyphen. A join key that is right in two spellings joins NOTHING, and it fails
/// silently — every company reads as a company with no LEI. So the normalisation is
/// part of the collector's declaration rather than something a downstream query has to
/// remember, and BOTH spellings land: the source's own (evidence) and the normalised
/// one (the key).
#[derive(Debug, Clone)]
pub struct Derive {
/// The column to add.
pub column: String,
/// The already-mapped column it is computed from.
pub from: String,
/// How.
pub how: DeriveHow,
}
/// **Page-NUMBER pagination** — the second of the two shapes a paginated JSON API has.
///
/// [`RestCollector::with_pagination`] follows a cursor the response CARRIES
/// (`links.next`), which is the JSON:API shape GLEIF serves. Domstolsverket serves the
/// other one: no cursor anywhere in the body, an `x-total-count` header, and a `page`
/// query parameter the caller counts up itself. A collector that could only follow a
/// cursor would land page zero of that source — 100 of 17 340 publications (measured
/// live 2026-09-05) — and report success.
///
/// **The stop condition is a SHORT page, not an empty one.** Asking for the page after
/// the last one costs an extra round trip on every single run, forever, and a source
/// that answers a 500 rather than an empty array to an over-the-end page would turn
/// that extra request into a failed run. A page that came back with fewer records than
/// `page_size` is already proof the server has no more.
#[derive(Debug, Clone)]
pub struct PageWalk {
/// The query parameter carrying the page number (`page`).
pub param: String,
/// The number the FIRST page has. `0` for Domstolsverket, `1` for a 1-based API.
/// The first request omits the parameter entirely and lets the server default;
/// this is the number that request is understood to have asked for.
pub first: u32,
/// How many records a FULL page holds — the stop condition. This must be the size
/// the server actually serves, not the size that was asked for: Domstolsverket
/// silently CLAMPS `pagesize` to 100 (measured 2026-09-05: `pagesize=500` and
/// `pagesize=1000` return byte-identical 100-record responses), and a walk that
/// believed it had asked for 500 would call the very first page short and stop.
pub page_size: usize,
}
/// How a [`Derive`] computes its value.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeriveHow {
/// Keep only ASCII digits. A NULL or empty source stays NULL — an all-punctuation
/// value would otherwise become the empty string and every such row would key
/// alike.
DigitsOnly,
}
/// **GLEIF's Swedish LEI slice, as one url** — filter, page size and all.
///
/// The filter name is the thing to get right and the thing the lane note got wrong:
/// `filter[entity.registeredAt.id]=RA000544` answers **HTTP 400** *"Filter should
/// contain only allowed values"*, while `filter[entity.legalAddress.country]=SE`
/// answers `meta.pagination.total` = **119 764** (measured 2026-09-04). Percent-encoded
/// here because the brackets are query syntax to some proxies and a literal to GLEIF.
///
/// `page[size]=200` is GLEIF's documented maximum, which is why
/// [`RestCollector::with_pagination`] had to exist before this collector could — and
/// also why it does not get the whole slice. See [`GLEIF_MAX_PAGES`].
pub const GLEIF_SWEDEN_URL: &str = "https://api.gleif.org/api/v1/lei-records\
?filter%5Bentity.legalAddress.country%5D=SE&page%5Bsize%5D=200";
/// **GLEIF refuses page-based pagination past 10 000 results, so this crawl reaches
/// 50 pages and no more.**
///
/// Measured 2026-09-04, and it is the single most important fact about this collector:
///
/// * `page[size]=200&page[number]=50` → **200**, `pagination.from` 9801 `to` 10000.
/// * `page[number]=51` → **400** *"Deep page-based pagination is not supported beyond
/// 10000 results (the product of page[size] and page[number] must not exceed 10000)"*.
///
/// So of Sweden's **119 764** LEI records this endpoint can serve **10 000 — 8.4 %**.
/// The cap is set to exactly the reachable page count rather than left at
/// [`DEFAULT_MAX_PAGES`], because hitting GLEIF's own 400 would fail the whole run and
/// land nothing, while stopping at 50 lands the 10 000 and reports that the table is
/// incomplete. A partial table that says it is partial is useful; a failed run is not,
/// and a partial table that claims completeness is worst of all — for a JOIN KEY it
/// would mean 91.6 % of Swedish companies silently reading as companies with no LEI.
///
/// **The full slice needs the Golden Copy bulk file, not this endpoint.** GLEIF says so
/// in its own response (`meta.goldenCopy.publishDate`), and
/// `https://goldencopy.gleif.org/api/v2/golden-copies/publishes` is live and keyless:
/// measured 2026-09-04, one publish per day at 08:00, `lei2` full file **3 421 669
/// records** as a 478 MB CSV zip (886 MB JSON), at a dated stable url. That is a
/// [`FileSeries`](crate::freshness::Freshness::FileSeries) collector — the publishes
/// listing IS the listing — and it is the next row, not a variation of this one.
pub const GLEIF_MAX_PAGES: usize = 50;
/// **Riksdagen's document list, as one url** — JSON, the 200-record maximum page, and
/// newest first.
///
/// `sz` caps at **200**: `sz=500` returns 200 (measured 2026-09-05), so asking for more
/// is not an error, it is a silently smaller page. `sort=datum&sortorder=desc` makes the
/// crawl read newest-first, which is what makes the 50-page ceiling land the RECENT
/// 10 000 rather than an arbitrary 10 000.
///
/// `doktyp=prop` narrows to the 31 812 propositioner. See
/// [`RestCollector::riksdagen_dokument`] for what the other doktyp values hold and why
/// the unfiltered 670 547 is not the default.
pub const RIKSDAGEN_DOKUMENTLISTA_URL: &str = "https://data.riksdagen.se/dokumentlista/\
?utformat=json&doktyp=prop&sz=200&sort=datum&sortorder=desc";
/// **Riksdagen addresses 10 000 results and then REPEATS, so this crawl stops at 50
/// pages.**
///
/// 50 × `sz=200` = the 10 000th result, and page 51 onwards is measurably the same page
/// as page 50's neighbourhood served over and over with HTTP 200 — the full measurement
/// is on [`RestCollector::riksdagen_dokument`], because that is where an operator reads
/// it. It is deliberately NOT [`DEFAULT_MAX_PAGES`]: a 700-page crawl here would land
/// ~130 000 rows of which ~129 800 are copies of one page, and report success.
pub const RIKSDAGEN_MAX_PAGES: usize = 50;
/// **The default page ceiling for a cursor crawl.**
///
/// 700 pages. GLEIF's Swedish slice is 119 764 records at 200 a page = **599** pages
/// (measured live 2026-09-04), so this is that crawl plus room for the register to grow
/// — not a round number, a measured one plus headroom. A crawl that reaches it lands
/// what it read and SAYS it stopped early, so the ceiling can never quietly truncate a
/// table an operator believes is complete.
pub const DEFAULT_MAX_PAGES: usize = 700;
impl RestCollector {
/// **Follow a cursor to the next page**, by RFC-6901 pointer into the response
/// (`/links/next` for JSON:API). Without this a paginated source lands its first
/// page and reports success.
pub fn with_pagination(mut self, next_pointer: &str) -> Self {
self.next_pointer = Some(next_pointer.into());
self
}
/// Override [`DEFAULT_MAX_PAGES`] for this preset.
pub fn with_max_pages(mut self, n: usize) -> Self {
self.max_pages = n.max(1);
self
}
/// **Walk page NUMBERS** for a source that carries no cursor — see [`PageWalk`].
/// Mutually exclusive with [`with_pagination`](RestCollector::with_pagination);
/// a cursor, when the source serves one, is always the better answer because it
/// survives the source reordering itself mid-crawl.
pub fn with_page_walk(mut self, param: &str, first: u32, page_size: usize) -> Self {
self.page_walk = Some(PageWalk {
param: param.into(),
first,
page_size: page_size.max(1),
});
self
}
/// **Drop a record the crawl has already read**, identified by RFC-6901 pointer —
/// see [`dedupe_pointer`](RestCollector::dedupe_pointer). The count of what it
/// dropped is REPORTED in `CollectStats::reason`, because a silent dedupe hides the
/// other half of an unstable sort: the records that were skipped.
pub fn with_dedupe_by(mut self, pointer: &str) -> Self {
self.dedupe_pointer = Some(pointer.into());
self
}
/// **The url of the page after this one, or `None` when the crawl is over.**
///
/// ONE writer for both pagination shapes, so the crawl loop in
/// [`collect`](Collector::collect) does not need a paginated twin of itself.
/// `first_url` is the url the FIRST request used (never the previous page's), so
/// the page parameter is appended to a clean base rather than stacked.
fn next_page(
&self,
resp: &Value,
first_url: &str,
pages_read: usize,
last_len: usize,
) -> Option<String> {
// A cursor the response carries always wins: it is the source's own answer.
if let Some(np) = &self.next_pointer {
return next_link(resp, np);
}
let pw = self.page_walk.as_ref()?;
// A SHORT page is the last page — see `PageWalk`.
if last_len < pw.page_size {
return None;
}
let n = pw.first as usize + pages_read;
let sep = if first_url.contains('?') { '&' } else { '?' };
Some(format!("{first_url}{sep}{}={n}", pw.param))
}
/// **Derive a normalised column from a mapped one** — see [`Derive`]. The source's
/// own spelling is kept alongside it.
pub fn with_derived(mut self, column: &str, from: &str, how: DeriveHow) -> Self {
self.derived.push(Derive {
column: column.into(),
from: from.into(),
how,
});
self
}
/// The field map (column ← pointer) this adapter applies — exposed so the
/// Ingest tab / tests can introspect the real adapter config.
pub fn fields(&self) -> &[FieldMap] {
&self.fields
}
/// The RFC-6901 pointer to the records array (`""` = whole doc).
pub fn records_pointer(&self) -> &str {
&self.records_pointer
}
/// **The page-number walk this preset declares**, if any — see [`PageWalk`].
///
/// Exposed because a test that builds the walk itself proves the MECHANISM and not
/// the PRESET, and that distinction is not academic: the first version of
/// `tests/domstol_rattspraxis.rs` re-added the walk in its own helper, so removing
/// `page_walk` from the shipped preset left all nine tests green over a collector
/// that would have landed 100 of 17 340 publications in production. Found by
/// driving the red arm (LAW 2), not by reading.
pub fn page_walk(&self) -> Option<&PageWalk> {
self.page_walk.as_ref()
}
/// The RFC-6901 pointer this preset deduplicates on across pages, if any — see
/// [`dedupe_pointer`](RestCollector::dedupe_pointer). Exposed for the same reason
/// [`page_walk`](RestCollector::page_walk) is.
pub fn dedupe_pointer(&self) -> Option<&str> {
self.dedupe_pointer.as_deref()
}
/// The declared incremental datetime field, if any (the CDC key).
pub fn bounds_field(&self) -> Option<&str> {
self.ts_pointer.as_deref()
}
/// Declare the per-record datetime field (RFC-6901 pointer) the [`Bounds`]
/// CDC filter narrows by — wiring the incremental seam for this adapter.
pub fn with_bounds_field(mut self, ts_pointer: &str) -> Self {
self.ts_pointer = Some(ts_pointer.into());
self
}
/// **Declare how this source CHANGES** — the freshness override. Without it the
/// adapter derives a local record watermark from
/// [`bounds_field`](RestCollector::bounds_field), or
/// [`AlwaysFull`](crate::freshness::Freshness::AlwaysFull) when it has none.
pub fn with_freshness(mut self, f: crate::freshness::Freshness) -> Self {
self.freshness = Some(f);
self
}
/// **Declare what identifies a ROW across runs** — the business-key override.
/// Without it the adapter derives one from its published join keys.
pub fn with_business_key(mut self, k: crate::history::BusinessKey) -> Self {
self.business_key = Some(k);
self
}
/// **Constant, non-secret headers this preset MUST send**, by name.
///
/// Digitraffic mandates `Digitraffic-User: <identifying string>` and answers **502**
/// without it (measured 2026-09-03). It is not a credential — there is nothing to
/// store and nothing to leak — so it must NOT be a [`SecretReq`](crate::requires::SecretReq):
/// a keyless collector that reported a missing secret would sit at `NeedsConfig`
/// forever with no key for the operator to supply. It rides as a header instead,
/// composed over whatever auth the config carries via [`crate::http::WithHeaders`].
///
/// Empty for every other preset, which is what "no extra headers" honestly is.
pub fn required_headers(&self) -> Vec<(String, String)> {
match self.name.as_str() {
"digitraffic-ais" => vec![(
"Digitraffic-User".to_string(),
DIGITRAFFIC_USER.to_string(),
)],
_ => Vec::new(),
}
}
/// **The request rate this preset's source publishes**, in requests/second.
///
/// Digitraffic documents **60 requests/minute**, so it gets 1/s rather than the
/// shared default of 10/s. A documented cap that only lives in a comment is not a
/// cap; this one reaches [`crate::http::HttpCore::new`] and actually throttles.
pub fn max_per_sec(&self) -> u32 {
match self.name.as_str() {
"digitraffic-ais" => 1,
// GLEIF publishes NO rate-limit header and rate-limits anyway: the
// offline-sample sweep crawled to page 137 at 10/s and came back HTTP 429
// (measured 2026-09-04). Absent headers are not an absent limit.
"gleif-lei" => 1,
// Domstolsverket publishes no cap and enforced none over the probe that was
// run (12 back-to-back requests, ~8/s, all HTTP 200, no `retry-after` and
// no `x-ratelimit-*` header — 2026-09-05). 1/s anyway: a 174-page crawl of a
// single agency's court register is not the place to find the limit
// experimentally, and GLEIF is the standing proof that an unpublished cap
// is still a cap.
"domstol-rattspraxis" => 1,
_ => 10,
}
}
/// **The human-facing card for this preset** — see [`crate::human_doc`].
///
/// One `match` on the preset name, the same shape [`RestCollector::max_per_sec`]
/// and [`RestCollector::required_headers`] already use, because the preset name IS
/// the discriminant for everything else about a `RestCollector`.
///
/// Every value here is either published by the source, already recorded in this
/// file's own doc comments, or MEASURED against the live endpoint on the date
/// named. Where none of those apply the field is `Unknown` — a real answer that
/// keeps an operator from trusting a guess.
pub fn human_doc_card(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit, Registration};
Some(match self.name.as_str() {
"polisen-events" => HumanDoc::own(
"Polismyndigheten (the Swedish Police)",
"https://polisen.se/om-webbplatsen/oppna-data/",
"https://polisen.se/om-webbplatsen/oppna-data/",
)
.with_contact(Contact::Form("https://polisen.se/kontakt/".into()))
.with_rate_limit(RateLimit::NonePublished)
.with_quirk(
"`/datetime` is LOCAL Swedish time with no zone marker, so a lexical \
watermark parks on the last single-digit hour and silently drops the \
rest of the day — the preset declares WindowGranularity::Day for this \
reason",
)
.with_quirk(
"the feed is a rolling window of the most recent events, NOT an \
archive: a gap in collection is a gap in your table for ever",
)
.with_quirk(
"the declared join keys are `location_name` + `gps`, which identify a \
PLACE and not an event — the business key is the source's own `/id`",
),
"kolada-kpi" => HumanDoc::own(
"RKA — Rådet för främjande av kommunala analyser",
"https://www.kolada.se/",
"https://api.kolada.se/v3/",
)
.with_contact(Contact::Form("https://www.kolada.se/om-oss/kontakt/".into()))
.with_rate_limit(RateLimit::NonePublished)
.with_quirk(
"MEASURED 2026-09-03 and 09-04: the v2 API answers HTTP 410 Gone with \
`{\"error\": \"This endpoint is deprecated. Please use /v3 instead.\"}` \
— v3 serves the same `{\"values\":[…]}` envelope, and v1 is 404",
),
"smhi-metobs" | "smhi-weather" | "smhi-hydroobs" => HumanDoc::own(
"SMHI — Sveriges meteorologiska och hydrologiska institut",
"https://www.smhi.se/data/oppna-data/villkor-for-anvandning",
"https://opendata.smhi.se/apidocs/",
)
.with_contact(Contact::Form("https://www.smhi.se/kontakta-oss".into()))
.with_rate_limit(RateLimit::NonePublished)
.with_quirk(
"station-keyed series: a station that stops reporting simply stops \
appearing, it does not report a gap",
),
"digitraffic-ais" => HumanDoc::new(
"Fintraffic (the Digitraffic service)",
"Fintraffic — vessel reports received by the Finnish AIS network",
"https://www.digitraffic.fi/en/terms-of-service/",
"https://www.digitraffic.fi/en/marine-traffic/",
)
.with_contact(Contact::Form("https://www.digitraffic.fi/en/support/".into()))
.with_rate_limit(RateLimit::Documented("60 requests/minute (1/s), which `max_per_sec` enforces".into()))
.with_quirk(
"MEASURED 2026-09-03: a request WITHOUT `Accept-Encoding: gzip` is \
refused with HTTP 406 and the body `Use of gzip compression is \
required with Accept-Encoding: gzip header.` — reqwest needs its \
`gzip` feature or this collector cannot fetch at all",
)
.with_quirk(
"MEASURED 2026-09-03: no `Digitraffic-User` header gets HTTP 502, and a \
plain `User-Agent` does NOT substitute. It is an identification \
courtesy, not a credential, so it is a constant header and not a secret",
)
.with_quirk(
"`properties.timestamp` is the AIS SECOND-OF-MINUTE (0..59), not a \
time: `properties.timestampExternal` is the epoch-ms. Keying on the \
raw field collapses an hour of fixes into 60 rows",
)
.with_quirk(
"`/api/ais/v1/vessels` (the ship-name lookup) answered 503 while \
`/locations` served fine, so no `name` column is wired; 503s on this \
service recover on their own and should be treated as transient",
),
"aisstream" => HumanDoc::new(
"AISStream.io",
"AISStream.io — a community AIS aggregation feed",
"https://aisstream.io/",
"https://aisstream.io/documentation",
)
.with_contact(Contact::Unknown)
.with_registration(Registration::SelfService { url: "https://aisstream.io/authenticate".into() })
.with_rate_limit(RateLimit::Unknown)
.with_quirk(
"it reads `stream.aisstream.io` and always did — it was called \
`marinetraffic-ais` until 2026-09-04, and the vendor in that name was \
the only thing about it that ever said MarineTraffic",
)
.with_quirk("a WebSocket stream, so coverage depends on who is contributing receivers"),
"opensky-flights" | "opensky-europe-bbox" | "opensky-tracks" => HumanDoc::own(
"The OpenSky Network",
"https://opensky-network.org/about/terms-of-use",
"https://openskynetwork.github.io/opensky-api/",
)
.with_contact(Contact::Form("https://opensky-network.org/about/contact".into()))
.with_registration(Registration::SelfService {
url: "https://opensky-network.org/ — then Account ▸ API client for the 2025 Keycloak OAuth2 client id/secret".into(),
})
.with_rate_limit(RateLimit::Documented(
"a daily credit budget; anonymous access is heavily rate-limited and an \
authenticated client gets substantially more — see the API docs for \
the current numbers".into(),
))
.with_quirk(
"CC-BY-SA 4.0 and NON-COMMERCIAL / research use — the share-alike arm \
reaches anything you publish derived from it",
)
.with_quirk(
"the 2025 flow is OAuth2 client-credentials against Keycloak, NOT the \
old username/password basic auth that most examples on the web still show",
),
"gbfs-citybikes" => HumanDoc::new(
"whoever operates the GBFS feed you configure — GBFS is a SPECIFICATION, not a service",
"the bike-share operator that publishes the feed",
"https://gbfs.org/licensing/",
"https://gbfs.org/specification/reference/",
)
.with_contact(Contact::NonePublished)
.with_rate_limit(RateLimit::Unknown)
.with_quirk(
"there is no central GBFS server: the terms, the rate limit and the \
support channel are the CONFIGURED operator's, and korp cannot know \
them until you point it somewhere",
)
.with_quirk("`last_reported` is per-vehicle and stale entries linger — key on (bike_id, last_reported)"),
"osm-overpass" => HumanDoc::new(
"the Overpass API instance you configure (the public one is run by volunteers)",
"OpenStreetMap contributors",
"https://www.openstreetmap.org/copyright",
"https://wiki.openstreetmap.org/wiki/Overpass_API",
)
.with_contact(Contact::Form("https://wiki.openstreetmap.org/wiki/Overpass_API".into()))
.with_rate_limit(RateLimit::Documented(
"the public instances enforce per-IP slot and timeout limits and WILL \
refuse you — a large or repeated extract belongs on your own instance".into(),
))
.with_quirk(
"ODbL is SHARE-ALIKE: a derived database you publish carries the same \
licence. This is the licence obligation most likely to surprise",
)
.with_quirk("Overpass answers 429/504 under load rather than queueing — treat both as retry-later, not as no-data"),
"bolagsverket-organisationer" => HumanDoc::own(
"Bolagsverket",
"https://bolagsverket.se/omoss/etjanstermm/apier",
"https://portal.api.bolagsverket.se/",
)
.with_contact(Contact::Form("https://bolagsverket.se/omoss/kontaktaoss".into()))
.with_registration(Registration::SelfService {
url: "https://portal.api.bolagsverket.se/ — register an application there for BOLAGSVERKET_CLIENT_ID + BOLAGSVERKET_CLIENT_SECRET".into(),
})
.with_rate_limit(RateLimit::Unknown)
.with_quirk(
"MEASURED 2026-09-03: the OAuth2 token endpoint is NOT on the gateway — \
`gw.api.bolagsverket.se/oauth2/token` is 404 for both GET and POST. The \
live one is `portal.api.bolagsverket.se/oauth2/token`, and it reads the \
client id/secret from the FORM BODY",
)
.with_quirk(
"`POST /organisationer` is POST-only (a GET is 405) and takes ONE \
company per call: `{\"identitetsbeteckning\":\"<orgnr>\"}`",
)
.with_quirk(
"there is NO kommunkod and no län in this API. The preset used to \
declare `kommunkod` as a join key, which would have become an upsert \
key over a column that never lands",
)
.with_quirk(
"board membership is NOT structured in the årsredovisning, and \
konkurs/likvidation events are POIT (HTML only, no API)",
),
"scb-foretagsregistret" => HumanDoc::own(
"Statistics Sweden (SCB), Företagsregistret",
"https://www.scb.se/vara-tjanster/bestall-data-och-statistik/foretagsregistret/avgiftsfria-uppgifter-i-foretagsregistret/",
"https://www.scb.se/vara-tjanster/bestall-data-och-statistik/foretagsregistret/foretagsregistrets-tjanster/foretagsregistrets-webbtjanster/",
)
.with_contact(Contact::Email(SCB_FORETAGSREGISTRET_CONTACT.into()))
.with_registration(Registration::ByRequest {
how: "mail scbforetag@scb.se with the organisation name and orgnr, the \
name and address of whoever accepts the terms of use, the name, \
address and mobile number of whoever receives the credential, and \
which data layouts you want. Free of charge since 2025-06-26 — the \
wait is a human one, not a fee"
.into(),
})
.with_rate_limit(RateLimit::Unknown)
.with_quirk(
"THIS COLLECTOR CANNOT RUN and says so by name. `Requirements::readiness` \
reports two missing things — the endpoint (SCB publishes it nowhere; it \
comes with the credentials) and SCB_FORETAGSREGISTRET_KEY — so ▶ Collect \
and the daemon both refuse instead of collecting a 401",
)
.with_quirk(
"the AUTHORISATION IS MID-MIGRATION: SCB's own page says access is by \
certificate + password today and \"kommer att bytas från nuvarande \
certifikat till API-nyckel\" in September 2026, which is now. One secret \
is declared, matching the key regime; if a CERTIFICATE arrives instead \
this is still blocked, because `korp_collectors::http` has no client- \
certificate path (no Identity, no PKCS#12) and mutual TLS is a separate \
piece of work",
)
.with_quirk(
"the RESPONSE SHAPE is not published either — the variable layout is a \
per-subscriber postbeskrivning PDF issued with the credential. The field \
map is therefore EMPTY on purpose: a guessed one would land a table of \
nulls and pass every structural check in this crate",
)
.with_quirk(
"what it unblocks, precisely: every Swedish company and workplace with \
its SNI code, address and employee-count band, in bulk and keyed by \
organisationsnummer. Nothing korp has today reaches that — the PxWeb \
cube NV/NV0101/FDBR07N is national aggregates with no orgnr in it at \
all, Bolagsverket resolves one orgnr per POST (and waits on its own two \
credentials), and GLEIF's page-based API stops at 10 000 of Sweden's \
119 764 LEIs",
)
.with_quirk(
"free is not open: SCB dropped the fee on 2025-06-26 but access stays \
per-subscriber under terms of use accepted at issue, so this is not \
CC0 and its rows must not be redistributed as if they were",
),
"domstol-rattspraxis" => HumanDoc::own(
"Domstolsverket (the Swedish National Courts Administration)",
"https://www.domstol.se/om-webbplatsen-och-digitala-kanaler/oppna-data/",
DOMSTOL_OPENAPI_URL,
)
.with_contact(Contact::Email("rattsfallspublikation@dom.se".into()))
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::Unknown)
.with_quirk(
"MEASURED 2026-09-05 — the DEFAULT SORT SILENTLY LOSES ROWS. Sorted by \
`avgorandedatum` (a date, no tiebreak) a 7-page crawl of the same 610 \
publications returned 610 rows carrying 607 distinct ids: 3 repeated at \
page boundaries and 3 never returned at all. With \
`sortorder=publiceringstid` the same crawl returned 610 of 610, \
including all three the default sort dropped. This preset pins the sort \
AND deduplicates on `id`",
)
.with_quirk(
"the parameter names are NOT guessable and are documented only in \
`openapi/puh-openapi.yaml`. Ten plausible guesses (`size`, `limit`, \
`antal`, `pageSize`, `fran`, `fromDate`, `domstolKod`, …) were each \
measured to return HTTP 200 with the parameter silently IGNORED — the \
real ones are `page`, `pagesize`, `domstolkod`, `publiceringstyper`, \
`malnummer`, `publicerad_fran_och_med`, `sortorder`, `asc`",
)
.with_quirk(
"`pagesize` is CLAMPED to 100 without saying so: `pagesize=500` and \
`pagesize=1000` return byte-identical 100-record responses \
(2026-09-05). 17 340 publications is therefore 174 requests, minimum",
)
.with_quirk(
"there is no cursor of any kind in the body — the total lives only in \
the `x-total-count` RESPONSE HEADER, and the sibling `POST /api/v1/sok` \
is the only endpoint that returns it in a body (`{total, \
publiceringLista}`) while ignoring every paging parameter it is given",
)
.with_quirk(
"`innehall` (the full judgment text) is NOT landed: 258 of 610 carry it, \
median 18 944 characters — and one single judgment carries 24 552 235, \
which is 27.5 MB of that slice's 34.8 MB in ONE cell. The `bilaga` \
columns point at the PDF instead (`GET /api/v1/bilagor/{lagringId}`). \
Note the transfer still costs those bytes: the API has no way to ask \
for a record without its text",
)
.with_quirk(
"`ecliNummer` is declared by the OpenAPI and was present on 0 of 610 \
publications (2026-09-05). The column lands and is all-NULL — a source \
that declares a field and does not fill it is worth recording as such",
)
.with_quirk(
"coverage is two different eras and the API does not say which a row is \
in: rättsfallsreferat from 1981 onward, but full domar och beslut only \
from March 2025 (Domstolsverket's own statement)",
),
"gbif-occurrences" => HumanDoc::own(
"GBIF — the Global Biodiversity Information Facility",
"https://www.gbif.org/terms",
"https://techdocs.gbif.org/en/openapi/",
)
.with_contact(Contact::Email("helpdesk@gbif.org".into()))
.with_rate_limit(RateLimit::Documented(
"the search API is paged and capped at 300 records per request; a bulk \
need is a DOWNLOAD request, not a hammered search".into(),
))
.with_quirk(
"the licence is per-DATASET, not per-API: an occurrence search returns \
records under a mix of CC0, CC-BY and CC-BY-NC, and attribution is owed \
to each publishing dataset",
),
"gleif-lei" => HumanDoc::own(
"GLEIF — the Global Legal Entity Identifier Foundation (Basel, \
established by the FSB/G20)",
"https://www.gleif.org/en/meta/lei-data-terms-of-use",
"https://documentation.gleif.org/API/",
)
.with_contact(Contact::Email("info@gleif.org".into()))
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::MeasuredNotPublished(
"no `x-ratelimit-*` and no `retry-after` on a 200, and HTTP 429 arrived \
at roughly page 137 of a 10/s crawl on 2026-09-04 — this preset runs at \
1 req/s, which makes the 599-page Swedish slice a ~10-minute crawl that \
finishes"
.into(),
))
.with_quirk(
"the country filter is `filter[entity.legalAddress.country]`, NOT \
`filter[entity.registeredAt.id]` — the latter answers HTTP 400 \
\"Filter should contain only allowed values\" (measured 2026-09-04, and \
korp's own lane note had the wrong one recorded as verified)",
)
.with_quirk(
"`entity.registeredAs` is the organisationsnummer WITH the hyphen \
(`559586-2557`); Bolagsverket's `orgnr` carries none, so the landed \
table adds a digits-only `orgnr` column beside it — joining on the raw \
value matches nothing and reports nothing",
)
.with_quirk(
"`filter[registration.lastUpdateDate]` needs a CLOSED range \
(`from..to`); the open-ended `from..` answers HTTP 400 \"Date must not \
be empty.\" — so this collector re-reads from the top and SAYS so \
rather than claiming a delta it cannot ask for",
)
.with_quirk(
"THE BIG ONE: page-based paging stops at the 10 000th result \
(`page[size]*page[number] <= 10000`, HTTP 400 past it), so this \
endpoint serves 10 000 of Sweden's 119 764 LEIs — 8.4 %. The run \
reports the table INCOMPLETE. The whole slice needs the Golden Copy \
bulk file (goldencopy.gleif.org, keyless, one publish a day at 08:00, \
3 421 669 records as a 478 MB CSV zip — measured 2026-09-04)",
),
"riksdagen-dokument" => HumanDoc::own(
"Sveriges riksdag (Riksdagsförvaltningen)",
"https://www.riksdagen.se/sv/dokument-och-lagar/riksdagens-oppna-data/",
"https://www.riksdagen.se/sv/dokument-och-lagar/riksdagens-oppna-data/",
)
.with_contact(Contact::Form("https://www.riksdagen.se/sv/kontakt/".into()))
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::NonePublished)
.with_quirk(
"THE BIG ONE: the list addresses 10 000 results and then REPEATS instead \
of refusing. Measured 2026-09-05 with `sz=200&sort=datum&sortorder=desc`: \
pages 51, 159, 160, 161 and 200 all answer HTTP 200 with the SAME 200 \
rows (`G60314 … G50348`, 1981-10-22 … 1982-08-19, 200/200 overlap), \
`@traff_till` pinned at 10000, and a fresh `@nasta_sida` each time. So \
the crawl is capped at 50 pages and the run SAYS the table is \
incomplete: 10 000 of 31 812 propositioner, 31.4 %",
)
.with_quirk(
"`from=YYYY-MM-DD` is a real server-side lower bound on `datum` (`prop` \
from 2026-01-01 → 219 hits against 31 812 unbounded, measured \
2026-09-05), so after the first backfill the window keeps every run far \
under the 10 000 wall. `tom=` exists for an upper bound; korp's plan \
carries one bound and sends only `from`",
)
.with_quirk(
"`@nasta_sida` is served as `http://` and the plain-HTTP host answers 302 \
to `https://` — the crawl works only because the HTTP client follows \
redirects. On the last page the key is ABSENT rather than empty",
)
.with_quirk(
"`sz` caps at 200: `sz=500` silently returns 200 rows, not an error. And \
`undertitel`, `relaterat_id` and `tempbeteckning` are 0/200 populated on \
`doktyp=prop`, so they are deliberately not mapped — an always-NULL \
column is a column an operator has to learn to ignore",
)
.with_quirk(
"SFS lives HERE, not on regeringen.se: svenskforfattningssamling.se \
publishes no RSS or Atom feed at all (checked 2026-09-05 — `/rss`, \
`/feed`, `/sfs.rss` all 404, no `<link rel=alternate>` on the home \
page), while `doktyp=sfs` on this endpoint reports 11 556 documents. \
Change the one query parameter",
)
.with_quirk(
"UNMEASURED: the collector runs at the shared 10 req/s default and the \
deepest live crawl anyone has done through it is TWO pages (400 rows, \
1.76 MB, 2026-09-05). A full 50-page backfill at 10/s has never been \
run, and lane EC O4 learned on GLEIF that a source publishing no \
rate-limit header can still answer 429 — so watch the first backfill \
and drop `max_per_sec` if it does",
),
"worldbank-indicators" => HumanDoc::own(
"The World Bank",
"https://datacatalog.worldbank.org/public-licenses",
"https://datahelpdesk.worldbank.org/knowledgebase/topics/125589",
)
.with_contact(Contact::Form("https://datahelpdesk.worldbank.org/".into()))
.with_rate_limit(RateLimit::NonePublished)
.with_quirk(
"the response is a TWO-element array `[metadata, rows]`, so the records \
pointer is `/1` and a reader that takes the whole body lands one row \
of envelope",
)
.with_quirk("a missing observation is present with a null value rather than absent — count nulls, not rows"),
"restcountries" => HumanDoc::own(
"RESTCountries — a community-run open-source project, not a government body",
"https://gitlab.com/restcountries/restcountries/-/blob/master/LICENSE",
"https://restcountries.com/",
)
.with_contact(Contact::Form("https://gitlab.com/restcountries/restcountries/-/issues".into()))
.with_rate_limit(RateLimit::NonePublished)
.with_quirk(
"a volunteer project on free hosting: it has had outages and a paid \
tier proposal. Reference data, not a dependency to schedule against",
),
"metno-forecast" => HumanDoc::own(
"MET Norway — Meteorologisk institutt",
"https://api.met.no/doc/TermsOfService",
"https://api.met.no/weatherapi/locationforecast/2.0/documentation",
)
.with_contact(Contact::Form("https://api.met.no/doc/TermsOfService".into()))
.with_rate_limit(RateLimit::Documented(
"MET's terms require a throttled, identified client and ask you to \
honour `Expires` / `If-Modified-Since` rather than poll".into(),
))
.with_quirk(
"MET REFUSES a request that does not carry an identifying User-Agent \
with a contact in it — this is stated in their terms and is the single \
most common reason a first call fails",
)
.with_quirk("coordinates are truncated to 4 decimals by the service; asking for more precision does not get it"),
"mobility-database" | "mobility-database-rt" => HumanDoc::own(
"MobilityData",
"https://mobilitydatabase.org/terms-and-conditions",
"https://mobilitydatabase.org/api-docs",
)
.with_contact(Contact::Form("https://mobilitydatabase.org/".into()))
.with_registration(Registration::SelfService {
url: "https://mobilitydatabase.org/ — sign up for a refresh token, which is exchanged for a short-lived access token".into(),
})
.with_rate_limit(RateLimit::Unknown)
.with_quirk(
"the token you are given is a REFRESH token: it must be exchanged for an \
access token, and the access token expires — a config holding the \
access token directly will start failing without changing",
)
.with_quirk("it is a CATALOGUE of feeds; the feeds themselves carry their own operators' terms"),
"eea-wise-water" => HumanDoc::own(
"EEA — the European Environment Agency",
"https://www.eea.europa.eu/en/legal-notice",
"https://discodata.eea.europa.eu/",
)
.with_contact(Contact::Form("https://www.eea.europa.eu/en/about/contact-us".into()))
.with_rate_limit(RateLimit::Unknown)
.with_quirk(
"DiscoData is a SQL-over-HTTP endpoint: the query is in the URL, so a \
change to the SELECT is a change to the source identity and resets the \
read mark by construction",
)
.with_quirk("`TOP n` is required by the backend — an unbounded SELECT is refused rather than streamed"),
"destatis-genesis" => HumanDoc::own(
"Statistisches Bundesamt (Destatis) — GENESIS-Online",
"https://www.destatis.de/EN/Service/Impressum/copyright.html",
"https://www-genesis.destatis.de/datenbank/online",
)
.with_contact(Contact::Form("https://www.destatis.de/EN/Service/Contact/contact.html".into()))
.with_registration(Registration::SelfService {
url: "https://www-genesis.destatis.de/ — a free GENESIS-Online account, whose credentials the API reads".into(),
})
.with_rate_limit(RateLimit::Documented(
"GENESIS limits concurrent jobs per account and returns a job handle for \
a large table rather than the table".into(),
))
.with_quirk(
"DL-DE-BY-2.0 requires naming Destatis and the year of the dataset in \
anything you publish",
),
"cbs-odata" => HumanDoc::own(
"CBS — Centraal Bureau voor de Statistiek (Statistics Netherlands)",
"https://www.cbs.nl/en-gb/about-us/website/copyright",
"https://www.cbs.nl/en-gb/our-services/open-data-statline-as-open-data",
)
.with_contact(Contact::Form("https://www.cbs.nl/en-gb/about-us/contact".into()))
.with_rate_limit(RateLimit::NonePublished)
.with_quirk(
"OData v3/v4 paging: the response carries `odata.nextLink` and a reader \
that stops at the first page silently lands a prefix of the table",
),
// No card for a preset this match does not name. `None` is the honest
// answer and `tests/human_docs.rs` turns it into a NAMED red rather than a
// blank panel an operator finds on their own.
_ => return None,
})
}
/// **The auth this preset uses when the caller did not override it.**
///
/// Same shape as [`crate::gribnc`]'s `auth_config` (L5 — one pattern, not two): the
/// host's known-source roster may hand down a `cfg.auth`, but a collector run
/// through the fallback path (`CollectorCfg::new(req.source, …)`, which korp builds
/// for any collector absent from that roster) would otherwise get
/// [`AuthConfig::None`] and 401. The credentials themselves stay in env; only the
/// VARIABLE NAMES are declared here.
pub fn default_auth(&self) -> AuthConfig {
match self.name.as_str() {
"bolagsverket-organisationer" => AuthConfig::OAuth2ClientCredentials {
token_url: BOLAGSVERKET_TOKEN_URL.to_string(),
id_env: BOLAGSVERKET_CLIENT_ID_ENV.to_string(),
secret_env: BOLAGSVERKET_CLIENT_SECRET_ENV.to_string(),
},
_ => AuthConfig::None,
}
}
/// **The JSON body this preset POSTs**, or `None` for the ordinary conditional GET.
///
/// Bolagsverket's `/organisationer` is POST-only (`GET` → **405**, measured
/// 2026-09-03) and takes `{"identitetsbeteckning": "<orgnr>"}` — one organisation
/// per call, per its published OpenAPI. The orgnr comes from `params`, which
/// [`requirements`](Collector::requirements) declares as a required field, so the
/// error below is the belt to readiness' braces and never the first thing an
/// operator sees.
pub fn request_body(&self, cfg: &CollectorCfg) -> Result<Option<String>> {
match self.name.as_str() {
"bolagsverket-organisationer" => {
let id = cfg
.params
.get("identitetsbeteckning")
.and_then(|v| v.as_str())
.filter(|s| !s.trim().is_empty())
.ok_or_else(|| {
anyhow!(
"`bolagsverket-organisationer` needs `params.identitetsbeteckning` \
(the organisationsnummer to look up) — POST /organisationer \
resolves ONE identity per request"
)
})?;
Ok(Some(
serde_json::json!({ "identitetsbeteckning": id }).to_string(),
))
}
_ => Ok(None),
}
}
fn rest_schema(&self) -> Schema {
let mut fdefs: Vec<(&str, DataType, bool)> = self
.fields
.iter()
.map(|f| (f.column.as_str(), f.ty.data_type(), true))
.collect();
// **The DERIVED columns are columns of the landed table**, so the schema must
// declare them. It did not, and the graph-mapping sweep caught it immediately:
// `gleif_lei`'s `HAS_LEI` edge reads `orgnr`, which `collect()` lands and
// `schema()` had never heard of. A schema that under-describes what lands is
// the same class of lie as a business key naming a column that is not there.
fdefs.extend(
self.derived
.iter()
.map(|d| (d.column.as_str(), DataType::Utf8, true)),
);
schema(&fdefs)
}
/// **Polisen** recent events — `https://polisen.se/api/events` (top-level array).
///
/// # Freshness, measured on the live feed (2026-09-03)
///
/// The endpoint serves **no `ETag` and no `Last-Modified`**, so a conditional fetch
/// is impossible here — this is a **query window**, not a mutable document. It does
/// take `?DateTime=`, at DAY granularity: the unbounded feed is the last ~500
/// events (190 839 bytes measured), while `?DateTime=2026-09-02` is that one day
/// (24 715 bytes) — 7.7× less over the wire.
///
/// The per-record key is `/datetime`, stamped `2026-09-03 1:23:41 +02:00` — a
/// **single-digit hour**, so it is NOT lexically sortable and the order is
/// [`MarkOrder::Timestamp`](crate::freshness::MarkOrder::Timestamp). A lexical
/// watermark here parks on the last single-digit hour and drops the rest of the
/// day, silently.
///
/// The business key is `/id`, the source's own event id — narrower and more honest
/// than the declared join keys (`location_name`, `gps`), which are a geography and
/// identify a place, not an event.
pub fn polisen() -> Self {
Self {
name: "polisen-events".into(),
description: "Polisen incident notices (events) — no auth, no personal data.".into(),
license: "open".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["location_name".into(), "gps".into()],
ts_pointer: Some("/datetime".into()),
freshness: Some(crate::freshness::Freshness::query_window(
"DateTime",
"/datetime",
crate::freshness::MarkOrder::Timestamp,
crate::freshness::WindowGranularity::Day,
)),
business_key: Some(crate::history::BusinessKey::of(
&["id"],
"the event id polisen assigns — the join keys are a place, not an event",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "".into(),
fields: vec![
FieldMap::new("id", "/id", FieldType::I64),
FieldMap::new("datetime", "/datetime", FieldType::Str),
FieldMap::new("name", "/name", FieldType::Str),
FieldMap::new("type", "/type", FieldType::Str),
FieldMap::new("summary", "/summary", FieldType::Str),
FieldMap::new("url", "/url", FieldType::Str),
FieldMap::new("location_name", "/location/name", FieldType::Str),
FieldMap::new("gps", "/location/gps", FieldType::Str),
],
}
}
/// **Kolada** KPI catalogue — `https://api.kolada.se/v3/kpi` (`values` array).
///
/// # v2 is dead, and the server says where it went
///
/// Measured live 2026-09-03 (t14s):
///
/// * `GET https://api.kolada.se/v2/kpi?per_page=2` → **HTTP 410**, 65 B, body
/// verbatim: `{"error": "This endpoint is deprecated. Please use /v3 instead."}`
/// * `GET https://api.kolada.se/v3/kpi?per_page=2` → **HTTP 200**, 955 B
/// * `GET https://api.kolada.se/v1/kpi?per_page=2` → HTTP 404
///
/// The v3 payload is the SAME envelope — `{"values":[{"id":"N00003","title":…,
/// "publication_date":…}, …]}` — so `records_pointer` (`/values`), the field map and
/// the join keys (`id`, `kommunkod`) are unchanged and **only the URL moved**. Keep
/// the collector, repoint it.
///
/// A stale v2 literal anywhere in the crate is the shape that rots silently: the
/// freshness MARK is keyed on `(collector, url)`, so a leftover v2 URL would make
/// the first v3 run look like a different source (or, worse, keep asking a 410).
/// `no_kolada_v2_url_survives_anywhere` in this module's tests is the guard.
pub fn kolada() -> Self {
Self {
name: "kolada-kpi".into(),
description: "Kolada KPI metadata (RKA) — ~6000 municipal/regional KPIs.".into(),
license: "open".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["id".into(), "kommunkod".into()],
ts_pointer: None,
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["id"],
"`/v3/kpi` is the KPI CATALOGUE — one row per indicator, and `kommunkod` is a join axis of the VALUES endpoint, not a column of this one",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/values".into(),
fields: vec![
FieldMap::new("id", "/id", FieldType::Str),
FieldMap::new("title", "/title", FieldType::Str),
FieldMap::new("operating_area", "/operating_area", FieldType::Str),
FieldMap::new("description", "/description", FieldType::Str),
],
}
}
/// **SMHI** metobs latest values — station-set JSON (`value` array).
pub fn smhi_metobs() -> Self {
Self {
name: "smhi-metobs".into(),
description: "SMHI meteorological observations (metobs) — latest values per station."
.into(),
license: "CC-BY 4.0".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["coordinate".into()],
ts_pointer: None,
freshness: Some(crate::freshness::Freshness::MutableDocument {
// measured live 2026-09-03: serves `last-modified`, and a second request carrying `If-Modified-Since` came back HTTP/2 304 with no body. No ETag.
validators: vec![crate::freshness::Validator::LastModified],
}),
business_key: Some(crate::history::BusinessKey::append_only(
"the station and the parameter are URL segments, never columns: the landed rows are `(date, value, quality)` and two stations' readings for the same instant are indistinguishable. Emitting `station_key` + `parameter` is what would give these rows an identity",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/value".into(),
fields: vec![
FieldMap::new("date", "/date", FieldType::I64),
FieldMap::new("value", "/value", FieldType::F64),
FieldMap::new("quality", "/quality", FieldType::Str),
],
}
}
/// **Digitraffic AIS** ship positions — Fintraffic's open marine AIS feed at
/// `https://meri.digitraffic.fi/api/ais/v1/locations`. Lands a **time-stamped
/// position table** (mmsi, lon, lat, sog, cog, heading, timestamp) so
/// facett-map3d can overlay + ANIMATE ship tracks and time-travel them via skade
/// AS-OF.
///
/// # This replaced a URL that never existed
///
/// Until 2026-09-03 this preset was `ais-positions` pointing at
/// `https://ais-feed.example/positions` — an RFC 2606 placeholder. It read as
/// `NeedsConfig` forever, `DataLayer::Boats` had no live source at any price, and
/// nothing on the box could have fetched a single vessel. Digitraffic is the
/// keyless replacement, and it is MEASURED, not assumed.
///
/// # Measured live 2026-09-03 (t14s)
///
/// * `GET /api/ais/v1/locations` with `Digitraffic-User` + `Accept-Encoding: gzip`
/// → **HTTP 200**, `application/json;charset=UTF-8`, **40 400 B on the wire**,
/// **417 567 B inflated**, **1001 features**.
/// * Without `Accept-Encoding: gzip` → **HTTP 406**, 70 B: `Use of gzip compression
/// is required with Accept-Encoding: gzip header.` (hence the crate's `gzip`
/// reqwest feature — see `Cargo.toml`).
/// * Without `Digitraffic-User` → **HTTP 502** (the gateway refuses an
/// unidentified client; a plain `User-Agent` does NOT substitute).
/// * `GET /api/ais/v1/vessels` (the static vessel-metadata sibling, name/callsign/
/// ship type) → **HTTP 503** at the same moment, so no name column is wired here:
/// this table carries the identity the feed actually serves, the MMSI.
///
/// # The record, and why `timestamp` is the EXTERNAL one
///
/// Records are GeoJSON `Feature`s under `/features`; the position is
/// `geometry.coordinates` as `[lon, lat]`. `properties.timestamp` is the raw AIS
/// field — a **second-of-minute, 0..59**, useless as a key — while
/// `properties.timestampExternal` is Fintraffic's epoch-millisecond receive time.
/// So the business-key column `timestamp` is bound to `timestampExternal`, and the
/// AIS second-of-minute lands beside it as `ts_second` rather than being dropped.
/// Keying `(mmsi, timestamp)` on the 0..59 field would collapse 60 fixes an hour
/// into 60 rows forever.
///
/// Keyless, but rate-limited to **60 requests/minute** — see
/// [`max_per_sec`](RestCollector::max_per_sec), which throttles this preset to 1/s
/// rather than the shared default of 10/s.
pub fn digitraffic_ais() -> Self {
Self {
name: "digitraffic-ais".into(),
description:
"Fintraffic Digitraffic AIS vessel locations (keyless, Digitraffic-User header) \
→ time-stamped ship position table (map overlay/animation)."
.into(),
license: "CC-BY 4.0 (Fintraffic / Digitraffic)".into(),
gdpr: GdprClass::Open,
// Keyless: no account, no key, no agreement. The mandatory
// `Digitraffic-User` header is an IDENTIFICATION header, not a credential,
// and rides via `required_headers()` so it never reads as a missing secret.
auth: AuthKind::None,
join_keys: vec!["mmsi".into(), "coordinate".into(), "timestamp".into()],
// The per-record time is an epoch INT (`timestampExternal`), which is not a
// lexically orderable CDC key — so this is a full snapshot per poll, exactly
// like the OpenSky `/states` feed. Row-level dedup is the business key's job.
ts_pointer: None,
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["mmsi", "timestamp"],
"one AIS position fix per vessel per receive time — `mmsi` is the vessel's \
permanent identity and `timestamp` is Digitraffic's epoch-ms \
`timestampExternal` (NOT the 0..59 AIS second-of-minute)",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/features".into(),
fields: vec![
FieldMap::new("mmsi", "/mmsi", FieldType::I64),
FieldMap::new("lon", "/geometry/coordinates/0", FieldType::F64),
FieldMap::new("lat", "/geometry/coordinates/1", FieldType::F64),
FieldMap::new("sog", "/properties/sog", FieldType::F64),
FieldMap::new("cog", "/properties/cog", FieldType::F64),
FieldMap::new("heading", "/properties/heading", FieldType::I64),
FieldMap::new("nav_stat", "/properties/navStat", FieldType::I64),
FieldMap::new("timestamp", "/properties/timestampExternal", FieldType::I64),
FieldMap::new("ts_second", "/properties/timestamp", FieldType::I64),
],
}
}
/// **OpenSky Network** live aircraft states — `https://opensky-network.org/api/states/all`.
/// Records live at `/states` as POSITIONAL arrays (icao24, callsign,
/// origin_country, time_position, last_contact, longitude, latitude,
/// baro_altitude, on_ground, velocity, true_track, …), so the field pointers
/// are numeric indices (`/0`, `/1`, …). Lands a position table facett-map3d
/// overlays + ANIMATES as flight tracks — the "flygplan" collector (boats =
/// AIS, energy = ENTSO-E). Anonymous access is rate-limited; no key required.
pub fn opensky() -> Self {
Self {
name: "opensky-flights".into(),
description:
"OpenSky Network live aircraft states (ADS-B open flight recorder) → time-stamped \
flight position table (map overlay/animation). Live auth: OAuth2 \
client-credentials (2025 Keycloak migration) via AuthConfig::OAuth2ClientCredentials \
reading OPENSKY_CLIENT_ID/OPENSKY_CLIENT_SECRET; anonymous is heavily rate-limited."
.into(),
license: "CC-BY-SA 4.0 (OpenSky Network; non-commercial/research)".into(),
gdpr: GdprClass::Open,
// 2025: authenticated quota needs an OAuth2 client-credentials token
// (anonymous/basic-auth 401s past the free rate limit).
auth: AuthKind::Token,
join_keys: vec!["icao24".into(), "callsign".into()],
// A live states snapshot has no clean lexical CDC key (per-record time
// is an epoch int) → full snapshot per poll, like the AIS position feed.
ts_pointer: None,
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["icao24"],
"a CURRENT-STATE table: one row per aircraft, upserted to its latest state vector. `callsign` is state, not identity — it changes mid-flight and is null on the ground, so keying on it would fork one aircraft into several rows",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/states".into(),
fields: vec![
FieldMap::new("icao24", "/0", FieldType::Str),
FieldMap::new("callsign", "/1", FieldType::Str),
FieldMap::new("origin_country", "/2", FieldType::Str),
FieldMap::new("lon", "/5", FieldType::F64),
FieldMap::new("lat", "/6", FieldType::F64),
FieldMap::new("baro_altitude", "/7", FieldType::F64),
FieldMap::new("velocity", "/9", FieldType::F64),
FieldMap::new("true_track", "/10", FieldType::F64),
FieldMap::new("geo_altitude", "/13", FieldType::F64),
],
}
}
/// **OpenSky Network** live aircraft states, **bounded to a Europe bbox** —
/// `https://opensky-network.org/api/states/all?lamin=34.5&lomin=-25.0&lamax=72.0&lomax=45.0`.
/// Same POSITIONAL-array `/states` shape as [`RestCollector::opensky`], but
/// the server returns ONLY aircraft inside the bounding box, landing a far
/// smaller, region-scoped flight position table for facett-map3d to overlay
/// + ANIMATE. The bbox rides the endpoint query string (via `base_url` at run
/// time); defaults roughly cover Europe (lat 34.5..72.0, lon -25.0..45.0).
pub fn opensky_europe_bbox() -> Self {
Self {
name: "opensky-europe-bbox".into(),
description:
"OpenSky Network live aircraft states bounded to a Europe bbox (lamin/lomin/lamax/lomax) \
→ time-stamped flight position table (map overlay/animation)."
.into(),
license: "CC-BY-SA 4.0 (OpenSky Network; non-commercial/research)".into(),
gdpr: GdprClass::Open,
// Same 2025 OAuth2 client-credentials requirement as `opensky`.
auth: AuthKind::Token,
join_keys: vec!["icao24".into(), "callsign".into()],
ts_pointer: None,
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["icao24"],
"a CURRENT-STATE table: one row per aircraft, upserted to its latest state vector. `callsign` is state, not identity — it changes mid-flight and is null on the ground",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/states".into(),
fields: vec![
FieldMap::new("icao24", "/0", FieldType::Str),
FieldMap::new("callsign", "/1", FieldType::Str),
FieldMap::new("origin_country", "/2", FieldType::Str),
FieldMap::new("lon", "/5", FieldType::F64),
FieldMap::new("lat", "/6", FieldType::F64),
FieldMap::new("baro_altitude", "/7", FieldType::F64),
FieldMap::new("velocity", "/9", FieldType::F64),
FieldMap::new("true_track", "/10", FieldType::F64),
FieldMap::new("geo_altitude", "/13", FieldType::F64),
],
}
}
/// **OpenSky Network** flight TRACK — the RECORDED trajectory of ONE aircraft
/// over time (`https://opensky-network.org/api/tracks/all?icao24={icao24}&time=0`).
/// This is the "flight recorder" sense — distinct from the instantaneous
/// [`RestCollector::opensky`] state-vector snapshot: the `/tracks` document
/// carries a `path[]` of ORDERED waypoints, each a POSITIONAL array
/// `[time, latitude, longitude, baro_altitude, true_track, on_ground]`, so the
/// field pointers are numeric indices (`/0`, `/1`, …) like the states arm. The
/// waypoints arrive time-ascending (the aviation twin of a saga tape); the
/// generic adapter preserves that order. Lands an `opensky_tracks` table
/// facett-map3d overlays + REPLAYS as a flight path. `/tracks` needs auth
/// (historically HTTP basic auth via `OPENSKY_USER`/`OPENSKY_PASS`).
pub fn opensky_tracks() -> Self {
Self {
name: "opensky-tracks".into(),
description:
"OpenSky Network recorded flight TRACK (flight recorder) → ordered (time, lat, lon, \
baro_altitude, heading, on_ground) waypoint trajectory for one aircraft → \
map overlay/replay. The time-ordered path sibling of the opensky-flights state \
snapshot. Live `/tracks` auth is credential-gated (OPENSKY_USER/OPENSKY_PASS)."
.into(),
license: "CC-BY-SA 4.0 (OpenSky Network; non-commercial/research)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::Token,
join_keys: vec!["icao24".into(), "callsign".into(), "time".into()],
// The waypoint `time` is a per-record epoch int, not a lexical string →
// no clean CDC bounds key; a track is fetched whole per aircraft.
ts_pointer: None,
freshness: None,
business_key: Some(crate::history::BusinessKey::append_only(
"the aircraft is an URL parameter (`icao24=`), never a column: the landed waypoints are `(time, lat, lon, …)` and two aircraft's tracks are indistinguishable once landed. Emitting `icao24` is what would give a waypoint an identity",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/path".into(),
fields: vec![
FieldMap::new("time", "/0", FieldType::I64),
FieldMap::new("lat", "/1", FieldType::F64),
FieldMap::new("lon", "/2", FieldType::F64),
FieldMap::new("baro_altitude", "/3", FieldType::F64),
FieldMap::new("heading", "/4", FieldType::F64),
FieldMap::new("on_ground", "/5", FieldType::Str),
],
}
}
/// **OSM Overpass API** features — a SELECTABLE Overpass QL query run against
/// a public Overpass endpoint (`https://overpass-api.de/api/interpreter`).
/// Overpass takes the query as `?data=` on a GET, so the query lives entirely
/// in `base_url` and the SAME adapter serves ANY node selector without a code
/// change. Records live at `/elements`; this arm lands the POINT features
/// (nodes): `lat`/`lon` are the coordinate, `name` is `tags/name`, and the
/// whole `tags` object is serialized to a Utf8 JSON string (the osm.rs `tags`
/// shape). Lands a point feature table facett-map3d overlays as a POI layer.
pub fn osm_overpass() -> Self {
Self {
name: "osm-overpass".into(),
description:
"OSM Overpass API features (selectable Overpass QL query) → point feature table (map POI overlay)."
.into(),
license: "ODbL (OpenStreetMap)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["coordinate".into(), "osm_id".into()],
ts_pointer: None,
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["osm_id", "osm_type"],
"an OSM id is unique only WITHIN its element type — node 12345 and way 12345 are different objects, so the type is part of the identity, not decoration",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/elements".into(),
fields: vec![
FieldMap::new("osm_id", "/id", FieldType::I64),
FieldMap::new("osm_type", "/type", FieldType::Str),
FieldMap::new("lat", "/lat", FieldType::F64),
FieldMap::new("lon", "/lon", FieldType::F64),
FieldMap::new("name", "/tags/name", FieldType::Str),
// The full tags object → compact JSON string (the osm.rs tags shape).
FieldMap::new("tags", "/tags", FieldType::Str),
],
}
}
/// **SMHI** open weather observation STATIONS → a point table (station_key,
/// name, owner, lat, lon, height, active) a facett map scene renders as
/// clickable points. Records live at `/station` in a metobs parameter
/// station-set (`.../parameter/1.json` = air temperature). Geometry sibling
/// of [`RestCollector::smhi_metobs`] (which lands the `/value` series); the
/// values join back via `station_key`. Static points → the `osm3d` scene.
pub fn smhi_weather() -> Self {
Self {
name: "smhi-weather".into(),
description:
"SMHI open weather observation stations → point table (station geometry for a map scene)."
.into(),
license: "CC-BY 4.0".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["station_key".into(), "coordinate".into()],
ts_pointer: None,
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["station_key"],
"the station roster — `station_key` is SMHI's own permanent station id, and one row is one station",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/station".into(),
fields: vec![
FieldMap::new("station_key", "/key", FieldType::Str),
FieldMap::new("name", "/name", FieldType::Str),
FieldMap::new("owner", "/owner", FieldType::Str),
FieldMap::new("lat", "/latitude", FieldType::F64),
FieldMap::new("lon", "/longitude", FieldType::F64),
FieldMap::new("height", "/height", FieldType::F64),
FieldMap::new("active", "/active", FieldType::Str),
],
}
}
/// **GBFS city-bikes** — a General Bikeshare Feed Specification
/// `free_bike_status.json` feed (dockless vehicle positions). Records live at
/// `/data/bikes` as objects carrying WGS84 coords, so this lands a
/// time-stamped position table (bike_id, lat, lon, is_reserved, is_disabled,
/// vehicle_type_id, current_range_meters) facett-map3d overlays + animates —
/// the bike sibling of AIS/OpenSky/TrainPosition. `last_updated` is a
/// document-level epoch int (not a per-record key) → full snapshot per poll.
/// GBFS v1.0/v2.0 encode `is_reserved`/`is_disabled` as 0/1 ints (I64 here).
pub fn gbfs_citybikes() -> Self {
Self {
name: "gbfs-citybikes".into(),
description:
"GBFS free_bike_status (city-bikes) → time-stamped position table (map overlay/animation)."
.into(),
license: "open (GBFS; feed-dependent, typically CC-BY/CC0)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None, // GBFS status feeds are public, no key
join_keys: vec!["bike_id".into(), "coordinate".into()],
ts_pointer: None,
freshness: Some(crate::freshness::Freshness::MutableDocument {
// measured live 2026-09-03: serves BOTH `etag` and `last-modified` — a GBFS status feed is the textbook mutable document, rewritten in place every minute or two.
validators: vec![crate::freshness::Validator::ETag, crate::freshness::Validator::LastModified],
}),
business_key: Some(crate::history::BusinessKey::of(
&["bike_id"],
"one row per vehicle — `bike_id` is the operator's own permanent id; `coordinate` is where it is, which is exactly the thing that changes",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/data/bikes".into(),
fields: vec![
FieldMap::new("bike_id", "/bike_id", FieldType::Str),
FieldMap::new("lat", "/lat", FieldType::F64),
FieldMap::new("lon", "/lon", FieldType::F64),
FieldMap::new("is_reserved", "/is_reserved", FieldType::I64),
FieldMap::new("is_disabled", "/is_disabled", FieldType::I64),
FieldMap::new("vehicle_type_id", "/vehicle_type_id", FieldType::Str),
FieldMap::new("current_range_meters", "/current_range_meters", FieldType::F64),
],
}
}
/// **AISStream.io** ship positions — the open aisstream live feed, batched to a
/// JSON dump of `PositionReport` messages at `/messages`. Each message is NESTED:
/// vessel identity + fix in `MetaData`, kinematics in `Message/PositionReport` — so
/// the field pointers are nested paths. Lands a time-stamped ship position table
/// facett-map3d overlays + ANIMATES; `time_utc` is the CDC bounds key. Needs a free
/// key (`AISSTREAM_API_KEY`).
///
/// # The name was a lie until 2026-09-03
///
/// This preset was called `marinetraffic-ais`. It has read `stream.aisstream.io`
/// since the day it was written, has never read MarineTraffic, parses AISStream's
/// own `MetaData`/`Message` envelope, and asks for an **AISStream** key. A name
/// that points at the wrong vendor is a wrong answer to "where did this row come
/// from?", so it is now `aisstream` — the operator, the licence and the key all
/// agree with the name.
pub fn aisstream() -> Self {
Self {
name: "aisstream".into(),
description: "AISStream.io open AIS PositionReport feed → time-stamped ship position table (map overlay/animation)."
.into(),
license: "open (AISStream.io free tier; feed-dependent)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::ApiKey, // AISStream requires a free self-serve API key
join_keys: vec!["mmsi".into(), "coordinate".into(), "timestamp".into()],
// Per-message ISO-8601 UTC string → lexically sortable CDC bounds key.
ts_pointer: Some("/MetaData/time_utc".into()),
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["mmsi", "timestamp"],
"one AIS position fix per vessel per report time — `timestamp` is \
AISStream's ISO-8601 `MetaData/time_utc`",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/messages".into(),
fields: vec![
FieldMap::new("mmsi", "/MetaData/MMSI", FieldType::I64),
FieldMap::new("name", "/MetaData/ShipName", FieldType::Str),
FieldMap::new("lat", "/MetaData/latitude", FieldType::F64),
FieldMap::new("lon", "/MetaData/longitude", FieldType::F64),
FieldMap::new("sog", "/Message/PositionReport/Sog", FieldType::F64),
FieldMap::new("cog", "/Message/PositionReport/Cog", FieldType::F64),
FieldMap::new("heading", "/Message/PositionReport/TrueHeading", FieldType::I64),
FieldMap::new("nav_status", "/Message/PositionReport/NavigationalStatus", FieldType::I64),
FieldMap::new("timestamp", "/MetaData/time_utc", FieldType::Str),
],
}
}
/// **Bolagsverket Värdefulla datamängder** — Swedish company/organisation master
/// data keyed by **organisationsnummer**, the entity-resolution spine every
/// business dataset (GLEIF LEI, FI, TED, Skatteverket aggregates) joins on.
///
/// # This repointed a dead URL onto the FREE API
///
/// Until 2026-09-03 this preset pointed at
/// `.../naringslivsregistret/v1/organisationer` with [`AuthKind::Agreement`] — an
/// endpoint that does not resolve, behind a paid agreement nobody here has signed.
/// The **Värdefulla datamängder** ("high-value datasets", the EU HVD obligation) API
/// is Bolagsverket's free, self-service replacement: register an application on the
/// developer portal, get an OAuth2 client id/secret, no fee and no signed agreement.
///
/// # Measured live 2026-09-03 (t14s) — the shape is not assumed
///
/// The gateway is WSO2 and answers unauthenticated probes with the shape, so every
/// line below is a measurement, not a guess:
///
/// | request | result |
/// |---|---|
/// | `POST /vardefulla-datamangder/v1/organisationer` | **401** `{"code":"900902","message":"Missing Credentials"}` |
/// | `GET` the same path | **405** — it is POST-only |
/// | `POST /vardefulla-datamangder/v1/dokumentlista` | **401**; `GET` → 405 |
/// | `GET /vardefulla-datamangder/v1/dokument/{id}` | **401**; `POST` → 405 |
/// | `GET /vardefulla-datamangder/v1/isalive` | **401** |
///
/// The published OpenAPI 3.0.3 document (devportal API `VärdefullaDatamängder` v1,
/// 74 379 B, fetched 2026-09-03) declares exactly those four operations, `servers[0]
/// = https://gw.api.bolagsverket.se/vardefulla-datamangder/v1`, a
/// `clientCredentials` flow with scope `vardefulla-datamangder:read`, the request
/// body `{"identitetsbeteckning": "<orgnr>"}` (a single string — ONE organisation per
/// call, not a bulk dump) and the response envelope `{"organisationer": [ … ]}`
/// whose member shape is what the field pointers below are taken from, verbatim.
///
/// # The token endpoint is NOT on the gateway host
///
/// The OpenAPI declares `tokenUrl: "/oauth2/token"`, which relative to the gateway
/// origin would be `https://gw.api.bolagsverket.se/oauth2/token`. **That URL is
/// dead**: measured `404 {"code":"404", … "The requested resource is not available."}`
/// for GET and POST, as are `/token` and `/authorize` on the same host. The live
/// token endpoint is on the PORTAL host — `https://portal.api.bolagsverket.se/oauth2/token`,
/// measured `401 {"error":"invalid_client","error_description":"A valid OAuth client
/// could not be found for client_id: probe"}` with `WWW-Authenticate: Basic
/// realm=portal.api.bolagsverket.se`. That error proves two things korp needs: the
/// endpoint is live, and it **reads client id/secret out of the FORM BODY** (it
/// looked our fake `client_id` up), which is exactly what the shared
/// [`oauth2_client_credentials_token`](crate::http::oauth2_client_credentials_token)
/// already sends — no second OAuth flow, no basic-auth variant.
///
/// # No credentials on this box, and it must SAY so
///
/// `BOLAGSVERKET_CLIENT_ID` / `BOLAGSVERKET_CLIENT_SECRET` are not registered yet.
/// [`requirements`](Collector::requirements) declares both by name plus the
/// `identitetsbeteckning` param, so `Requirements::readiness` refuses the run —
/// naming all three — on both the ▶ Collect and the daemon path, instead of a 401
/// arriving at the wire as a parse failure.
///
/// # What this API does NOT carry
///
/// There is no `kommunkod` and no `län` in the response — the seat is a postal
/// address (`postnummer`/`postort`). So the join keys are `orgnr` alone; the
/// kommun join is korp's own `AddressIndex` problem, not something to declare here
/// and never fill.
/// **GLEIF LEI records — Sweden's slice, and the JOIN BRIDGE korp was missing.**
///
/// Lane EC O4. Without this table an orgnr and a LEI are two unrelated identifiers,
/// so **FI's insider register** (`fi-insyn`, which names the issuer only by LEI) and
/// **TED procurement notices** do not join to a Swedish company at all. This is the
/// row that makes them one graph.
///
/// # Everything below was measured against the live API on 2026-09-04
///
/// **The filter name in the lane note was WRONG and the note said it was verified.**
/// `filter[entity.registeredAt.id]=RA000544` answers **HTTP 400**
/// `{"errors":[{"title":"Filter should contain only allowed values."}]}`. The filter
/// that works is `filter[entity.legalAddress.country]=SE`, and it answers
/// `meta.pagination.total` = **119 764** — so the count in the note was about right
/// and the parameter was not. Trust the source, not the note.
///
/// * `page[size]=200` is GLEIF's documented maximum — and **only the first 50 pages
/// are reachable**: `page[number]=51` answers 400 *"Deep page-based pagination is
/// not supported beyond 10000 results"*. So this endpoint serves **10 000 of the
/// 119 764 — 8.4 %** — and the run SAYS the table is incomplete. See
/// [`GLEIF_MAX_PAGES`] for the bulk path that gets the rest.
/// * `links.next` carries the next page url, absolute and already query-encoded;
/// the last page serves `links: {}` — see [`next_link`].
/// * `attributes.entity.registeredAs` is the **orgnr, hyphenated**: `559586-2557`
/// on the record this was read from. Bolagsverket's `orgnr` column carries no
/// hyphen, so a join on the raw value matches NOTHING and fails silently — every
/// company would read as a company with no LEI. Hence the `orgnr` derived column
/// ([`DeriveHow::DigitsOnly`]) beside the source's own `registered_as`.
/// * **GLEIF rate-limits and serves no header saying so.** `x-ratelimit-*` and
/// `retry-after` are all absent on a 200, which is what made the first version of
/// this preset run at the shared 10/s default — and the offline-sample sweep then
/// crawled to `page[number]=137` and came back **HTTP 429 Too Many Requests**
/// (measured 2026-09-04). Absent headers are not an absent limit. So this preset
/// declares **1 req/s**: 599 pages is then a ten-minute crawl that finishes,
/// rather than 136 pages and a wall.
///
/// # Why it is AlwaysFull today, and what would change that
///
/// `filter[registration.lastUpdateDate]` **exists** — the API names it back
/// (`"source":{"parameter":"filter.registration->lastUpdateDate"}`) — but it wants a
/// **closed range**: `<from>..<to>` returns rows (**292** SE records for
/// `2026-09-03..2026-09-05`), while the open-ended `<from>..` answers 400 `"Date must
/// not be empty."`. `Plan::Window` puts ONE value in one parameter, so a `from..to`
/// window needs a second bound the plan does not carry yet. Declaring a window korp
/// cannot fill would be worse than declaring the honest full crawl, so this says
/// `AlwaysFull` and the range is the next row.
pub fn gleif_lei() -> Self {
Self {
name: "gleif-lei".into(),
description:
"GLEIF Level 1 LEI records for Sweden (api.gleif.org v1, keyless, CC0) — the \
orgnr ↔ LEI bridge that lets FI's insider register and TED procurement \
notices join to a Swedish company."
.into(),
license: "CC0 1.0 (GLEIF publishes the LEI data free of charge and free of \
restriction)"
.into(),
gdpr: GdprClass::Open, // legal entities, not natural persons
auth: AuthKind::None,
join_keys: vec!["orgnr".into(), "lei".into()],
// No usable server-side window yet — see the doc comment. A declared
// `ts_pointer` here would make the SECOND run report "unchanged, 0 rows"
// over a source it never actually narrowed.
ts_pointer: None,
freshness: Some(crate::freshness::Freshness::AlwaysFull {
why: "GLEIF's `filter[registration.lastUpdateDate]` needs a CLOSED range \
(`from..to`) and `Plan::Window` carries one bound, so every run \
re-reads from the top — and page-based paging stops at the 10 000th \
result, so every run reads the SAME first 10 000 of Sweden's 119 764 \
(measured 2026-09-04). Both limits are named rather than dressed up \
as a delta; the Golden Copy bulk file is the way out of both."
.into(),
}),
business_key: Some(crate::history::BusinessKey::of(
&["lei"],
"one row per LEI — the LEI is ISO 17442 and permanent; a company can hold \
more than one over time, so the orgnr is the JOIN key and not the identity",
)),
next_pointer: Some("/links/next".into()),
page_walk: None,
dedupe_pointer: None,
// NOT the default: GLEIF answers 400 past the 10 000th result, so the cap
// is the reachable page count — see `GLEIF_MAX_PAGES`.
max_pages: GLEIF_MAX_PAGES,
derived: vec![Derive {
column: "orgnr".into(),
from: "registered_as".into(),
how: DeriveHow::DigitsOnly,
}],
last_read: std::sync::Mutex::new(None),
records_pointer: "/data".into(),
fields: vec![
FieldMap::new("lei", "/attributes/lei", FieldType::Str),
FieldMap::new("legal_name", "/attributes/entity/legalName/name", FieldType::Str),
// The orgnr AS GLEIF SPELLS IT. `orgnr` (digits only) is derived from
// this one, and both land: the source's own value is the evidence, the
// normalised one is the key.
FieldMap::new("registered_as", "/attributes/entity/registeredAs", FieldType::Str),
FieldMap::new("entity_status", "/attributes/entity/status", FieldType::Str),
FieldMap::new("legal_form", "/attributes/entity/legalForm/id", FieldType::Str),
FieldMap::new("city", "/attributes/entity/legalAddress/city", FieldType::Str),
FieldMap::new("postal_code", "/attributes/entity/legalAddress/postalCode", FieldType::Str),
FieldMap::new("country", "/attributes/entity/legalAddress/country", FieldType::Str),
FieldMap::new("registration_status", "/attributes/registration/status", FieldType::Str),
FieldMap::new("last_update", "/attributes/registration/lastUpdateDate", FieldType::Str),
FieldMap::new("managing_lou", "/attributes/registration/managingLou", FieldType::Str),
FieldMap::new("bic", "/attributes/bic/0", FieldType::Str),
],
}
}
pub fn bolagsverket() -> Self {
Self {
name: "bolagsverket-organisationer".into(),
description:
"Bolagsverket Värdefulla datamängder (free HVD API, OAuth2 client-credentials) — \
company/organisation master data keyed by organisationsnummer."
.into(),
license: "CC0 / EU high-value datasets (Bolagsverket Värdefulla datamängder, free)"
.into(),
gdpr: GdprClass::Open, // company registry (organisations, not natural persons)
auth: AuthKind::ApiKey,
// No kommunkod in the payload — see the doc comment. Declaring one would
// make `BusinessKey::derived` key on a column that never lands.
join_keys: vec!["orgnr".into()],
// POST /organisationer is a LOOKUP by identitetsbeteckning, not a window over
// a time series: there is nothing to narrow by, and a `registreringsdatum`
// watermark would make the second lookup of the same company report
// "unchanged, 0 rows" for ever.
ts_pointer: None,
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["orgnr"],
"one row per organisation — `organisationsidentitet.identitetsbeteckning` \
is Bolagsverket's own permanent identity for the company",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/organisationer".into(),
// Pointers taken verbatim from the published OpenAPI's own
// `organisationer-aktiebolag-svar` example (2026-09-03).
fields: vec![
FieldMap::new(
"orgnr",
"/organisationsidentitet/identitetsbeteckning",
FieldType::Str,
),
FieldMap::new(
"namn",
"/organisationsnamn/organisationsnamnLista/0/namn",
FieldType::Str,
),
FieldMap::new("foretagsform", "/organisationsform/kod", FieldType::Str),
FieldMap::new(
"foretagsform_text",
"/organisationsform/klartext",
FieldType::Str,
),
FieldMap::new("juridisk_form", "/juridiskForm/kod", FieldType::Str),
FieldMap::new("verksam", "/verksamOrganisation/kod", FieldType::Str),
FieldMap::new(
"registreringsdatum",
"/organisationsdatum/registreringsdatum",
FieldType::Str,
),
FieldMap::new(
"avregistreringsdatum",
"/avregistreradOrganisation/avregistreringsdatum",
FieldType::Str,
),
FieldMap::new(
"avregistreringsorsak",
"/avregistreringsorsak/klartext",
FieldType::Str,
),
FieldMap::new("sni_kod", "/naringsgrenOrganisation/sni/0/kod", FieldType::Str),
FieldMap::new(
"sni_text",
"/naringsgrenOrganisation/sni/0/klartext",
FieldType::Str,
),
FieldMap::new(
"verksamhetsbeskrivning",
"/verksamhetsbeskrivning/beskrivning",
FieldType::Str,
),
FieldMap::new(
"postnummer",
"/postadressOrganisation/postadress/postnummer",
FieldType::Str,
),
FieldMap::new(
"postort",
"/postadressOrganisation/postadress/postort",
FieldType::Str,
),
],
}
}
/// **Riksdagen's open document register** — `data.riksdagen.se/dokumentlista`, the
/// Swedish parliament's whole corpus behind one keyless JSON URL.
///
/// # What is actually there (measured live 2026-09-05)
///
/// `?utformat=json&sz=1` reports `@traffar` per `doktyp`:
///
/// | doktyp | documents |
/// |---|---|
/// | *(all)* | **670 547** |
/// | `mot` motioner | 257 961 |
/// | `bet` betänkanden | 74 780 |
/// | `prop` propositioner | **31 812** |
/// | `sfs` författningar | 11 556 |
///
/// `sfs` is worth naming: `svenskforfattningssamling.se` publishes **no** RSS or
/// Atom feed at all (checked 2026-09-05 — `/rss`, `/feed`, `/sfs.rss` all 404, the
/// home page carries no `<link rel="alternate">`), so this endpoint is the machine
/// route to SFS, not regeringen.se's syndication.
///
/// The base URL narrows to `doktyp=prop` because 670 547 documents is not a table
/// anyone asked for and the page ceiling below cannot reach it. Change one query
/// parameter for another slice; the mapping is the same for every `doktyp`.
///
/// # The cap is 50 pages, and it is not a round number
///
/// **Riksdagen refuses nothing past the 10 000th result — it repeats.** Measured
/// 2026-09-05 with `sz=200&sort=datum&sortorder=desc`:
///
/// * pages 1 and 2 are distinct (`HD03309…HD03120`, then `HD03116…HC03151`).
/// * pages **51, 159, 160, 161 and 200 are the SAME 200 rows** — `G60314 … G50348`,
/// 1981-10-22 … 1982-08-19, overlap 200/200 between each consecutive pair — and
/// every one of them answers **HTTP 200** with `@traff_till` pinned at `10000` and
/// a fresh `@nasta_sida` pointing one page further.
///
/// That is worse than GLEIF's wall, which at least answers HTTP 400: a crawl that
/// trusted the cursor here would run to [`DEFAULT_MAX_PAGES`] and land **the same
/// page 650 times**, ~130 000 rows of which ~129 800 are duplicates, with nothing
/// red anywhere. The `seen`-url guard in [`Collector::collect`] does not catch it
/// either — each repeat has a DIFFERENT url (`&p=N`) and identical content.
///
/// So the ceiling is exactly the reachable page count: 50 pages × 200 = the 10 000
/// results the API can actually address, **31.4 %** of the 31 812 propositioner, and
/// the run REPORTS that the table is incomplete.
///
/// A second belt sits behind it: `dedupe_pointer = /dok_id` (lane EC O6's mechanism,
/// reused rather than twinned). Inside the addressable range the paging IS a stable
/// total order — pages 1 and 2 share no `dok_id` — so it never fires in normal
/// operation; if the wall ever moves BELOW the ceiling it turns 50 pages of one
/// repeated page into the 200 distinct documents they actually are, and the run says
/// how many duplicates it dropped.
///
/// # …which is why the window matters
///
/// `from=YYYY-MM-DD` is a real server-side lower bound on `datum` (measured: `prop`
/// with `from=2026-01-01` → 219 hits, versus 31 812 unbounded), so after the first
/// backfill every run asks for a slice far under the 10 000 wall and the cap stops
/// being a limit at all. `tom=` exists too and `Plan::Window` carries one bound, so
/// only `from` is declared.
///
/// # Two traps in the cursor
///
/// * `@nasta_sida` is served as **`http://`**, and `http://data.riksdagen.se/…`
/// answers **302** to the `https://` twin (measured 2026-09-05). The crawl works
/// because `reqwest` follows redirects by default; a client configured
/// `redirect::Policy::none()` would read page 1 and report success.
/// * On the last page `@nasta_sida` is **absent**, not empty — verified against a
/// windowed query whose 219 hits fit in two pages. [`next_link`] treats both the
/// same, so the crawl terminates either way.
///
/// # Fields
///
/// `undertitel`, `relaterat_id` and `tempbeteckning` are **not** mapped: they are
/// **0/200** on page 1 of `doktyp=prop`, and a column that is always NULL is not
/// evidence, it is a column an operator has to learn to ignore.
///
/// The rest are measured over a real two-page collect (2026-09-05, **400 rows, 400
/// distinct `dok_id`, 1 764 171 bytes**):
///
/// | column | filled |
/// |---|---|
/// | `dok_id`, `doktyp`, `titel`, `rm`, `beteckning`, `datum`, `summary` | 400/400 |
/// | `organ` | **371/400 — 92.8 %** |
/// | `status` | 336/400 — 84 % |
///
/// `organ` is the correction worth writing down: on page 1 alone it read 200/200 and
/// this preset nearly shipped claiming it is always there. It is not, and the graph
/// mapping's `ISSUED_BY` edge therefore does not exist for ~7 % of documents — a
/// NULL endpoint the sink drops, which is the right behaviour and not a silent one
/// only because it is written here. `status` is mapped despite being partial: partial
/// is not absent, and it is not a key.
pub fn riksdagen_dokument() -> Self {
Self {
name: "riksdagen-dokument".into(),
description:
"Riksdagen dokumentlista (data.riksdagen.se, keyless) — the Swedish \
parliament's document register: propositioner, betänkanden, motioner, \
SFS. One row per document, keyed on Riksdagen's own `dok_id`, windowed \
by `from=` on the document date."
.into(),
license: "Riksdagen öppna data — free reuse, no registration, attribution \
requested (riksdagen.se/sv/dokument-och-lagar/riksdagens-oppna-data/)"
.into(),
// Public parliamentary documents. Named politicians appear in the TEXT of a
// motion, which is publication, not a personal-data register — but a
// `personlista` read would be a different answer and is deliberately not
// this collector.
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["dok_id".into()],
ts_pointer: Some("/datum".into()),
freshness: Some(crate::freshness::Freshness::query_window(
"from",
"/datum",
crate::freshness::MarkOrder::Timestamp,
crate::freshness::WindowGranularity::Day,
)),
business_key: Some(crate::history::BusinessKey::of(
&["dok_id"],
"one row per document — `dok_id` is Riksdagen's own permanent identifier \
(`HD03305`), unique across the whole corpus and stable when a document \
is re-published; `rm`+`beteckning` is the human citation and repeats \
across doktyp",
)),
next_pointer: Some("/dokumentlista/@nasta_sida".into()),
// Riksdagen serves a real cursor, so the page-NUMBER walk lane EC O6 built
// for Domstolsverket is not needed here.
page_walk: None,
// **The second belt against the repeat, and it is not paranoia.**
//
// The cap keeps the crawl inside the 10 000 results the API can address, and
// inside that range the paging IS a stable total order (pages 1 and 2 share
// no `dok_id`, measured 2026-09-05). But the wall's position is the server's
// to move, and if it ever moves below the ceiling the crawl walks straight
// into the repeat — the same 200 documents, up to 50 times, with nothing red.
// `dok_id` is the source's own permanent identifier, so the dedupe costs a
// BTreeSet of 10 000 strings and turns that failure from a corrupted table
// into a short one that says it is short. Driven red 2026-09-05: with this
// `None`, a 50-page crawl of one repeated page lands 100 rows of 2 documents.
dedupe_pointer: Some("/dok_id".into()),
max_pages: RIKSDAGEN_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/dokumentlista/dokument".into(),
fields: vec![
FieldMap::new("dok_id", "/dok_id", FieldType::Str),
FieldMap::new("doktyp", "/doktyp", FieldType::Str),
FieldMap::new("subtyp", "/subtyp", FieldType::Str),
FieldMap::new("titel", "/titel", FieldType::Str),
// The riksmöte (`2025/26`) and the number within it — together the
// citation a Swede would actually search for, `Prop. 2025/26:305`.
FieldMap::new("rm", "/rm", FieldType::Str),
FieldMap::new("beteckning", "/beteckning", FieldType::Str),
FieldMap::new("nummer", "/nummer", FieldType::Str),
// The department or utskott that issued it — the one relationship these
// rows carry, and the edge the graph mapping draws.
FieldMap::new("organ", "/organ", FieldType::Str),
FieldMap::new("datum", "/datum", FieldType::Str),
FieldMap::new("publicerad", "/publicerad", FieldType::Str),
FieldMap::new("systemdatum", "/systemdatum", FieldType::Str),
FieldMap::new("status", "/status", FieldType::Str),
FieldMap::new("kalla", "/kalla", FieldType::Str),
FieldMap::new("dokument_url_html", "/dokument_url_html", FieldType::Str),
FieldMap::new(
"dokumentstatus_url_xml",
"/dokumentstatus_url_xml",
FieldType::Str,
),
FieldMap::new("summary", "/summary", FieldType::Str),
],
}
}
/// **GBIF** species occurrences — the Global Biodiversity Information Facility
/// occurrence search API (`https://api.gbif.org/v1/occurrence/search`). A
/// public, no-auth JSON API (GBIF is the EU-anchored, Copenhagen-hosted open
/// biodiversity network). Records live at `/results`; each carries the taxon
/// (`scientificName`), the WGS84 fix (`decimalLatitude`/`decimalLongitude`),
/// country, dataset key and observation date — so this lands a time-stamped
/// **biodiversity occurrence point table** a facett map overlays as a species
/// POI layer (the "nature" ontology sibling of the transport overlays).
pub fn gbif() -> Self {
Self {
name: "gbif-occurrences".into(),
description:
"GBIF species occurrences (Global Biodiversity Information Facility) → biodiversity occurrence point table (map POI overlay)."
.into(),
license: "CC-BY / CC0 (GBIF; dataset-dependent)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None, // GBIF occurrence search is public, no key
join_keys: vec!["coordinate".into(), "country".into(), "gbif_key".into()],
// Per-record ISO-8601 eventDate → lexically sortable CDC bounds key.
ts_pointer: Some("/eventDate".into()),
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["gbif_key"],
"`gbif_key` is GBIF's own stable occurrence key",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/results".into(),
fields: vec![
FieldMap::new("gbif_key", "/key", FieldType::I64),
FieldMap::new("scientific_name", "/scientificName", FieldType::Str),
FieldMap::new("kingdom", "/kingdom", FieldType::Str),
FieldMap::new("lat", "/decimalLatitude", FieldType::F64),
FieldMap::new("lon", "/decimalLongitude", FieldType::F64),
FieldMap::new("country", "/country", FieldType::Str),
FieldMap::new("basis_of_record", "/basisOfRecord", FieldType::Str),
FieldMap::new("dataset_key", "/datasetKey", FieldType::Str),
FieldMap::new("event_date", "/eventDate", FieldType::Str),
],
}
}
/// **World Bank** development indicators — the World Bank Open Data indicator
/// API (`https://api.worldbank.org/v2/country/all/indicator/{code}?format=json`).
/// A public, no-auth API. The response is a **two-element array**: element 0 is
/// paging metadata, element 1 is the observation array — so `records_pointer`
/// is `/1`. Each observation carries the indicator + country labels (nested
/// `/indicator/value`, `/country/value`), the ISO3 country code (the join
/// spine), the year and the value → a per-`(country, year, indicator)`
/// statistics table.
pub fn worldbank() -> Self {
Self {
name: "worldbank-indicators".into(),
description:
"World Bank Open Data development indicators → per-(country, year, indicator) statistics table."
.into(),
license: "CC-BY 4.0 (World Bank Open Data)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None, // World Bank Open Data is public, no key
join_keys: vec!["countryiso3code".into(), "date".into(), "indicator_id".into()],
// The observation year is a per-record lexical CDC bounds key.
ts_pointer: Some("/date".into()),
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["indicator_id", "countryiso3code", "date"],
"one observation per (indicator, country, period)",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
// Element 1 of the [meta, data] envelope is the observation array.
records_pointer: "/1".into(),
fields: vec![
FieldMap::new("indicator_id", "/indicator/id", FieldType::Str),
FieldMap::new("indicator", "/indicator/value", FieldType::Str),
FieldMap::new("country", "/country/value", FieldType::Str),
FieldMap::new("countryiso3code", "/countryiso3code", FieldType::Str),
FieldMap::new("date", "/date", FieldType::Str),
FieldMap::new("value", "/value", FieldType::F64),
],
}
}
/// **REST Countries** — the RESTCountries reference API
/// (`https://restcountries.com/v3.1/all?fields=name,cca2,cca3,region,population,latlng`).
/// A public, no-auth JSON API returning a **top-level array** of countries
/// (`records_pointer == ""`). Lands a country reference table (common name,
/// ISO-3166 alpha-2/alpha-3 codes — the country **join spine** every other
/// dataset resolves against, region/subregion, population, centroid lat/lon).
pub fn restcountries() -> Self {
Self {
name: "restcountries".into(),
description:
"RESTCountries reference data → country reference table (ISO-3166 code join spine)."
.into(),
license: "MPL-2.0 (RESTCountries; open)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["cca2".into(), "cca3".into()],
ts_pointer: None,
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["cca3"],
"the ISO-3166-1 alpha-3 code — one row per country, and `cca2` is a second spelling of the same identity rather than part of it",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "".into(),
fields: vec![
FieldMap::new("name", "/name/common", FieldType::Str),
FieldMap::new("official_name", "/name/official", FieldType::Str),
FieldMap::new("cca2", "/cca2", FieldType::Str),
FieldMap::new("cca3", "/cca3", FieldType::Str),
FieldMap::new("region", "/region", FieldType::Str),
FieldMap::new("subregion", "/subregion", FieldType::Str),
FieldMap::new("population", "/population", FieldType::I64),
FieldMap::new("lat", "/latlng/0", FieldType::F64),
FieldMap::new("lon", "/latlng/1", FieldType::F64),
],
}
}
/// **SMHI hydrological observations** (WATER) — SMHI's open hydrology API
/// (`https://opendata-download-hydroobs.smhi.se/api/.../data.json`), which
/// shares the metobs station-set JSON shape (a `value` array of dated
/// readings). Lands a **water** time series (river discharge / water level /
/// stage per station) — the hydrology sibling of [`RestCollector::smhi_metobs`]
/// (weather). Public, no-auth, CC-BY.
pub fn smhi_hydro() -> Self {
Self {
name: "smhi-hydroobs".into(),
description:
"SMHI hydrological observations (WATER: river discharge / water level) — latest values per station."
.into(),
license: "CC-BY 4.0 (SMHI)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["coordinate".into()],
ts_pointer: None,
freshness: Some(crate::freshness::Freshness::MutableDocument {
// measured live 2026-09-03: serves `last-modified`. No ETag.
validators: vec![crate::freshness::Validator::LastModified],
}),
business_key: Some(crate::history::BusinessKey::append_only(
"the station and the parameter are URL segments, never columns: the landed rows are `(date, value, quality)` and two stations' readings for the same instant are indistinguishable. Emitting `station_key` + `parameter` is what would give these rows an identity",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/value".into(),
fields: vec![
FieldMap::new("date", "/date", FieldType::I64),
FieldMap::new("value", "/value", FieldType::F64),
FieldMap::new("quality", "/quality", FieldType::Str),
],
}
}
/// **Met.no Locationforecast** (WEATHER) — the Norwegian Meteorological
/// Institute point-forecast API
/// (`https://api.met.no/weatherapi/locationforecast/2.0/compact?lat=..&lon=..`).
/// The forecast is a `properties.timeseries[]` array; each step carries an ISO
/// time + the instant details (temperature, wind, humidity, pressure, cloud).
/// Lands a per-`(time)` point-forecast table. Public, no-auth (Met.no only
/// requires an identifying User-Agent — the shared HTTP core sends one). The
/// weather-forecast sibling of SMHI metobs (observations).
pub fn metno_forecast() -> Self {
Self {
name: "metno-forecast".into(),
description:
"Met.no Locationforecast point forecast (WEATHER) → per-time forecast table (temp, wind, humidity, pressure, cloud)."
.into(),
license: "CC-BY 4.0 / NLOD (MET Norway)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None, // no key; an identifying User-Agent is required (HttpCore sends one)
join_keys: vec!["time".into()],
// Each forecast step is stamped with an ISO-8601 time → lexical CDC key.
ts_pointer: Some("/time".into()),
freshness: Some(crate::freshness::Freshness::MutableDocument {
// measured live 2026-09-03: serves `last-modified`. A forecast is REWRITTEN — the same /time values get revised — so a record watermark would silently skip every revision. The document, not the row, is what changed.
validators: vec![crate::freshness::Validator::LastModified],
}),
business_key: Some(crate::history::BusinessKey::append_only(
"the location is a URL parameter (`lat=`/`lon=`), never a column: the landed rows are `(time, air_temperature, …)`, so keying on `time` alone would MERGE every location's forecast for the same hour into one row. Emitting the coordinate is what would give these rows an identity",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/properties/timeseries".into(),
fields: vec![
FieldMap::new("time", "/time", FieldType::Str),
FieldMap::new("air_temperature", "/data/instant/details/air_temperature", FieldType::F64),
FieldMap::new("wind_speed", "/data/instant/details/wind_speed", FieldType::F64),
FieldMap::new("wind_from_direction", "/data/instant/details/wind_from_direction", FieldType::F64),
FieldMap::new("relative_humidity", "/data/instant/details/relative_humidity", FieldType::F64),
FieldMap::new("air_pressure_at_sea_level", "/data/instant/details/air_pressure_at_sea_level", FieldType::F64),
FieldMap::new("cloud_area_fraction", "/data/instant/details/cloud_area_fraction", FieldType::F64),
],
}
}
/// **Mobility Database** meta-collector — `https://api.mobilitydatabase.org/v1/`
/// (CC0). The global catalogue of public-transit + micromobility feed records
/// that SUCCEEDS the retired TransitFeeds (shut down Dec 2025): 6000+ GTFS /
/// GTFS-RT / GBFS feeds keyed by a stable `mdb-*` id. This is a **meta**
/// collector — its typed rows are FEED DESCRIPTORS (id, data_type, provider,
/// country, producer + latest-dataset URLs) that auto-drive the crate's
/// EXISTING [`crate::gtfs`] / [`crate::gtfs_rt`] collectors: point one at each
/// `producer_url` / `latest_url` and the whole catalogue becomes ingestible.
///
/// Records live at the top-level array (`records_pointer == ""`) of a
/// `/v1/gtfs_feeds` (or `/v1/gtfs_rt_feeds` / `/v1/gbfs_feeds`) response; the
/// endpoint family is chosen via `base_url` at collect time so ONE adapter
/// serves every feed type. Auth is a bearer access token (env
/// `MOBILITY_DB_TOKEN`, resolved via [`AuthConfig::Bearer`]); never hardcoded.
///
/// [`AuthConfig::Bearer`]: super::AuthConfig::Bearer
pub fn mobility_database() -> Self {
Self {
name: "mobility-database".into(),
description:
"Mobility Database feed catalogue (CC0) → typed GTFS/GTFS-RT/GBFS feed-descriptor \
rows (id, data_type, provider, country, feed URLs) that auto-drive the GTFS \
collectors — the TransitFeeds successor."
.into(),
license: "CC0-1.0 (Mobility Database)".into(),
gdpr: GdprClass::Open,
// The catalogue API is gated by a free bearer access token.
auth: AuthKind::Token,
join_keys: vec!["feed_id".into(), "country".into()],
ts_pointer: None,
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["feed_id"],
"`feed_id` is the Mobility Database's own permanent feed id; `country` is an attribute of the feed, not part of what identifies it",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "".into(),
fields: vec![
FieldMap::new("feed_id", "/id", FieldType::Str),
FieldMap::new("data_type", "/data_type", FieldType::Str),
FieldMap::new("status", "/status", FieldType::Str),
FieldMap::new("provider", "/provider", FieldType::Str),
FieldMap::new("feed_name", "/feed_name", FieldType::Str),
FieldMap::new("country", "/locations/0/country_code", FieldType::Str),
FieldMap::new(
"subdivision",
"/locations/0/subdivision_name",
FieldType::Str,
),
FieldMap::new("municipality", "/locations/0/municipality", FieldType::Str),
FieldMap::new("producer_url", "/source_info/producer_url", FieldType::Str),
FieldMap::new("latest_url", "/latest_dataset/hosted_url", FieldType::Str),
],
}
}
/// **EEA WISE water quality** — the European Environment Agency's Water
/// Information System for Europe (WISE-SoE Waterbase) aggregated water-quality
/// data, served over the **Discodata** SQL REST API
/// (`https://discodata.eea.europa.eu/sql?query=<SQL>`). The `?query=` SELECT
/// rides `base_url` at collect time, so ONE adapter serves any WISE table.
/// Records live at `/results`; each carries the monitoring site, country,
/// water-body category, the measured determinand + mean value/unit, the
/// reference year and the WGS84 fix — landing a per-`(site, determinand, year)`
/// water-quality table (the environmental sibling of the SMHI hydro series).
/// Public, no-auth, CC-BY.
pub fn eea_wise() -> Self {
Self {
name: "eea-wise-water".into(),
description:
"EEA WISE water quality (Waterbase, via the Discodata SQL API) → per-(site, determinand, year) water-quality table."
.into(),
license: "CC-BY 4.0 (EEA / Copernicus)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None, // Discodata SQL is public, no key
join_keys: vec!["monitoring_site".into(), "country".into(), "coordinate".into()],
// The reference year is a per-record lexical CDC bounds key.
ts_pointer: Some("/phenomenonTimeReferenceYear".into()),
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["monitoring_site", "determinand", "year"],
"one aggregate per (site, substance, year) — `country` is an attribute of the site",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/results".into(),
fields: vec![
FieldMap::new("monitoring_site", "/monitoringSiteIdentifier", FieldType::Str),
FieldMap::new("country", "/countryCode", FieldType::Str),
FieldMap::new("water_body_category", "/parameterWaterBodyCategory", FieldType::Str),
FieldMap::new("determinand", "/observedPropertyDeterminandLabel", FieldType::Str),
FieldMap::new("mean_value", "/resultMeanValue", FieldType::F64),
FieldMap::new("unit", "/resultUom", FieldType::Str),
FieldMap::new("year", "/phenomenonTimeReferenceYear", FieldType::I64),
FieldMap::new("lat", "/lat", FieldType::F64),
FieldMap::new("lon", "/lon", FieldType::F64),
],
}
}
/// **Destatis GENESIS-Online** — the German Federal Statistical Office national
/// statistics REST API (`https://www-genesis.destatis.de/genesisWS/rest/2020/`).
/// This arm lands the **table catalogue** (`catalogue/tables`): a `List` of
/// statistical-table DESCRIPTORS (table code, human title, time coverage) that
/// index which GENESIS cubes exist — a meta-collector whose `code` rows drive a
/// downstream `data/table` pull (the national-stats sibling of the Mobility
/// Database feed catalogue). Auth is a free GENESIS account token (query
/// param, env `DESTATIS_GENESIS_TOKEN`, resolved via [`super::AuthConfig`]).
pub fn destatis_genesis() -> Self {
Self {
name: "destatis-genesis".into(),
description:
"Destatis GENESIS-Online table catalogue (German national statistics) → statistical-table descriptor rows (code, title, time coverage)."
.into(),
license: "DL-DE-BY-2.0 (Destatis GENESIS)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::Token, // a free GENESIS account token (query param)
join_keys: vec!["table_code".into()],
ts_pointer: None,
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["table_code"],
"the catalogue — one row per GENESIS table",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/List".into(),
fields: vec![
FieldMap::new("table_code", "/Code", FieldType::Str),
FieldMap::new("content", "/Content", FieldType::Str),
FieldMap::new("time", "/Time", FieldType::Str),
],
}
}
/// **CBS Open Data** (Statistics Netherlands) — the national statistics OData
/// API (`https://opendata.cbs.nl/ODataApi/odata/<table>/TypedDataSet`). OData
/// returns the typed observations under a `value` array; the table id + any
/// `$filter`/`$select` ride `base_url` at collect time, so ONE adapter serves
/// any CBS cube. The field map here is a representative population-cube shape
/// (`ID`, `Perioden`, region, a topic value); a custom cube overrides it via
/// [`RestCollector::custom`]. Public, no-auth, CC-BY.
pub fn cbs_odata() -> Self {
Self {
name: "cbs-odata".into(),
description:
"CBS Open Data (Statistics Netherlands) OData TypedDataSet → typed national-statistics observation rows."
.into(),
license: "CC-BY 4.0 (CBS / Statistics Netherlands)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None, // CBS OData is public, no key
join_keys: vec!["region".into(), "period".into()],
// Each observation is stamped with a CBS period code → lexical CDC key.
ts_pointer: Some("/Perioden".into()),
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["id"],
"OData's own row key; `(region, period)` is a join axis and is not unique once a cube carries more than one measure",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "/value".into(),
fields: vec![
FieldMap::new("id", "/ID", FieldType::I64),
FieldMap::new("period", "/Perioden", FieldType::Str),
FieldMap::new("region", "/RegioS", FieldType::Str),
FieldMap::new("value", "/TotaleBevolking_1", FieldType::F64),
],
}
}
/// **Mobility Database — GTFS-Realtime feed catalogue** — the RT-feed face of
/// the Mobility Database meta-collector (`api.mobilitydatabase.org/v1/gtfs_rt_feeds`,
/// CC0). Where [`RestCollector::mobility_database`] enumerates STATIC GTFS
/// feeds, this enumerates **GTFS-Realtime** feeds: each row is an RT feed
/// descriptor (id, `entity_type` — `vp`/`tu`/`sa` for vehicle-positions /
/// trip-updates / alerts, provider, country + the `.pb` feed URL) that
/// auto-drives the EXISTING [`crate::gtfs_rt`] protobuf reader — point the
/// GTFS-RT collector at each `latest_url` and the whole live-transit catalogue
/// becomes ingestible. Auth is the same free bearer token (env
/// `MOBILITY_DB_TOKEN`, resolved via [`super::AuthConfig::Bearer`]); never
/// hardcoded.
pub fn mobility_database_rt() -> Self {
Self {
name: "mobility-database-rt".into(),
description:
"Mobility Database GTFS-Realtime feed catalogue (CC0) → typed RT feed-descriptor \
rows (id, entity_type, provider, country, .pb feed URL) that auto-drive the \
gtfs-rt-vehicles protobuf reader."
.into(),
license: "CC0-1.0 (Mobility Database)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::Token,
join_keys: vec!["feed_id".into(), "country".into()],
ts_pointer: None,
freshness: None,
business_key: Some(crate::history::BusinessKey::of(
&["feed_id"],
"`feed_id` is the Mobility Database's own permanent feed id",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "".into(),
fields: vec![
FieldMap::new("feed_id", "/id", FieldType::Str),
FieldMap::new("data_type", "/data_type", FieldType::Str),
FieldMap::new("entity_type", "/entity_types/0", FieldType::Str),
FieldMap::new("status", "/status", FieldType::Str),
FieldMap::new("provider", "/provider", FieldType::Str),
FieldMap::new("feed_name", "/feed_name", FieldType::Str),
FieldMap::new("country", "/locations/0/country_code", FieldType::Str),
FieldMap::new(
"subdivision",
"/locations/0/subdivision_name",
FieldType::Str,
),
FieldMap::new("municipality", "/locations/0/municipality", FieldType::Str),
FieldMap::new("latest_url", "/source_info/producer_url", FieldType::Str),
],
}
}
/// **Domstolsverket rättspraxis** — every published Swedish court decision, summary
/// and preliminary-ruling request from the higher courts (lane EC O6, half A).
///
/// Keyless and open. Measured live 2026-09-05: **17 340** publications across
/// **9** courts in the 2026 slice's own answer (`x-total-count`), rättsfallsreferat
/// from 1981 and full judgments/decisions from March 2025 (Domstolsverket's own
/// coverage statement).
///
/// Three things about this source are not obvious and each one is a measurement:
///
/// 1. **The sort has to be pinned** — see [`DOMSTOL_RATTSPRAXIS_URL`]. The default
/// sort silently drops 3 of every 610 records read.
/// 2. **The paging carries no cursor.** No `links.next`, no `next` field, nothing in
/// the body at all — only an `x-total-count` header and a `page` parameter the
/// caller counts up. That is why [`PageWalk`] had to exist before this preset
/// could: the cursor crawl would have landed page zero and reported success.
/// 3. **The identity is `id`, not `malnummer` + `domstolkod`.** The lane brief named
/// the second pair, and it is measurably not unique: over the 610-publication
/// 2026 slice, `(malNummerLista, domstolKod)` yields **605** distinct values for
/// **610** publications — 5 case numbers each published twice by the same court
/// (`M 14236-25` MMOD, `B 1864-25` HDO, `Ä 9632-24` HDO, `F 3201-25` MMOD,
/// `ÖP 8558-25` MMOD), which is the ordinary case of a court publishing both a
/// decision and its referat. `id` is a UUID and was 610-for-610 distinct under
/// the pinned sort. A business key on the brief's pair would have merged two
/// genuinely different publications into one warehouse row.
///
/// **`innehall` — the full text — is deliberately NOT landed.** See
/// [`FieldType::TextLen`]: 258 of the 610 carry it, median 18 944 characters, and
/// one single judgment carries **24 552 235** — 27.5 MB of the slice's 34.8 MB in
/// one cell. `innehall_tecken` lands its size instead, and
/// `bilaga_fillagring_id` points at the PDF (`GET /api/v1/bilagor/{lagringId}`),
/// which is a file-series collector and not this row.
///
/// GDPR: [`GdprClass::Open`] is the wrong-feeling but correct answer — these are
/// PUBLISHED court decisions, which Domstolsverket publishes precisely because they
/// are public documents. `sammanfattning` is an editorial summary written for
/// publication. The full text that is not landed here is where names would be.
pub fn domstol_rattspraxis() -> Self {
Self {
name: "domstol-rattspraxis".into(),
description:
"Domstolsverket rättspraxis (PUH API, keyless) — every published judgment, \
decision, referat, notis and förhandsavgörande-begäran from Sweden's higher \
courts: 17 340 publications across 9 courts, measured live 2026-09-05."
.into(),
license: "open data, no registration (Domstolsverket öppna data; the agency \
publishes no SPDX tag for it, so this says what it IS rather than \
inventing a licence name)"
.into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
// `id` and `domstol_kod` are both emitted columns; `domstol_kod` is the
// codelist `GET /api/v1/domstolar` publishes, so it is a real join key and
// not a hopeful one.
join_keys: vec!["id".into(), "domstol_kod".into()],
// The server-side lower bound, so a re-run is cheaper in BYTES and not only
// in rows — see `freshness` just below.
ts_pointer: Some("/publiceringstid".into()),
freshness: Some(crate::freshness::Freshness::query_window(
"publicerad_fran_och_med",
"/publiceringstid",
MarkOrder::Timestamp,
crate::freshness::WindowGranularity::Day,
)),
business_key: Some(crate::history::BusinessKey::of(
&["id"],
"one row per PUBLICATION, keyed on the API's own uuid — measured \
610-of-610 distinct over the 2026 slice, where the brief's \
`malnummer + domstolkod` gave only 605 (a court publishing both a \
decision and its referat repeats the case number)",
)),
next_pointer: None,
// No cursor anywhere in the body — page numbers, 0-based, 100 a page.
page_walk: Some(PageWalk {
param: "page".into(),
first: 0,
page_size: DOMSTOL_PAGE_SIZE,
}),
// Belt AND braces: the pinned sort removed every observed duplicate, but
// `publiceringstid` is minute-granular and two publications in the same
// minute across a page boundary would put them straight back.
dedupe_pointer: Some("/id".into()),
// 17 340 publications at 100 a page is 174 pages; 250 leaves the register
// room to grow for years and still stops a runaway. A crawl that reaches it
// lands what it read and SAYS the table is incomplete.
max_pages: 250,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
// The document IS the array — no envelope on this endpoint. (The sibling
// `POST /api/v1/sok` wraps the same records in `{total, publiceringLista}`
// and is the only place the total appears in a BODY, but it ignores every
// paging parameter, so it cannot serve a crawl.)
records_pointer: "".into(),
fields: vec![
FieldMap::new("id", "/id", FieldType::Str),
// The publication GROUP: a decision and its referat share one. 607
// distinct over the 610-record slice.
FieldMap::new(
"grupp_korrelationsnummer",
"/gruppKorrelationsnummer",
FieldType::Str,
),
FieldMap::new("domstol_kod", "/domstol/domstolKod", FieldType::Str),
FieldMap::new("domstol_namn", "/domstol/domstolNamn", FieldType::Str),
// PREJUDIKAT · VAGLEDANDE_MEN_EJ_PREJUDICERANDE · EJ_VAGLEDANDE ·
// PROVNINGSTILLSTAND · FORHANDSAVGORANDE (all five observed).
FieldMap::new("typ", "/typ", FieldType::Str),
// DOM_ELLER_BESLUT · REFERAT · NOTIS — and NULL on 61 of 610, which the
// OpenAPI permits (it is not in `PubliceringDTO.required`).
FieldMap::new("publiceringsform", "/publiceringsform", FieldType::Str),
FieldMap::new("avgorandedatum", "/avgorandedatum", FieldType::Str),
FieldMap::new("publiceringstid", "/publiceringstid", FieldType::Str),
// Repeated fields, joined — see `FieldType::Joined`.
FieldMap::new("mal_nummer", "/malNummerLista", FieldType::Joined),
FieldMap::new("referat_nummer", "/referatNummerLista", FieldType::Joined),
FieldMap::new("rattsomrade", "/rattsomradeLista", FieldType::Joined),
FieldMap::new("nyckelord", "/nyckelordLista", FieldType::Joined),
FieldMap::new("forarbete", "/forarbeteLista", FieldType::Joined),
// The European Case Law Identifier. Declared by the OpenAPI and
// observed on **0 of 610** publications (2026-09-05) — the column lands
// all-NULL today, which is the honest record of a field the source
// declares and does not fill.
FieldMap::new("ecli_nummer", "/ecliNummer", FieldType::Str),
FieldMap::new("benamning", "/benamning", FieldType::Str),
// The editorial summary — present on 610 of 610, median 257 characters,
// widest 1 910. This is the searchable substance.
FieldMap::new("sammanfattning", "/sammanfattning", FieldType::Str),
// The full text's SIZE, not the text. See `FieldType::TextLen`.
FieldMap::new("innehall_tecken", "/innehall", FieldType::TextLen),
// The PDF. 324 of 610 carry exactly one attachment, 6 carry more; only
// the first is mapped, and the doc comment says so rather than the
// table pretending a publication has one document.
FieldMap::new(
"bilaga_fillagring_id",
"/bilagaLista/0/fillagringId",
FieldType::Str,
),
FieldMap::new("bilaga_filnamn", "/bilagaLista/0/filnamn", FieldType::Str),
],
}
}
/// **SCB Företagsregistret** — the per-organisationsnummer business register, and
/// the collector that CANNOT RUN (lane EC O6, half B).
///
/// This one exists to refuse correctly, by name, and to say what it would do. Every
/// other route to Swedish company data korp has is aggregate or partial:
///
/// * [`crate::pxweb::SCB_FDBR07N_URL`] answers SNI × storleksklass × år for the whole
/// country and contains no organisationsnummer at all — that is the CEILING of the
/// keyless half, not a temporary state.
/// * `bolagsverket-organisationer` resolves ONE orgnr per POST and is itself waiting
/// on two credentials.
/// * `gleif-lei` reaches 10 000 of Sweden's 119 764 LEI records.
///
/// The register behind this key is the only source that answers "every Swedish
/// company and workplace, with its SNI code, address and size band" in bulk. It is
/// **free** (SCB dropped the fee on 2025-06-26) and gated by a credential requested
/// by mail to [`SCB_FORETAGSREGISTRET_CONTACT`].
///
/// **The field map is deliberately EMPTY, and that is load-bearing.** SCB does not
/// publish the response shape either — the variable layout comes with the
/// credentials, in a *postbeskrivning* PDF chosen per subscriber. A field map
/// written from a guess would land a table of nulls and pass every structural test in
/// this crate. An empty one means `schema()` emits no columns, `business_key`
/// declares append-only, and nothing anywhere pretends to know what this table looks
/// like until someone has actually seen the layout.
pub fn scb_foretagsregistret() -> Self {
Self {
name: "scb-foretagsregistret".into(),
description:
"SCB Företagsregistret (the per-organisationsnummer business register, REST/JSON) \
— BLOCKED: needs the SCB_FORETAGSREGISTRET_KEY credential, requested from \
scbforetag@scb.se. Free since 2025-06-26 and still gated. Until it arrives, \
korp's SCB half is PxWeb aggregates only (SNI × storleksklass × år), never an \
orgnr."
.into(),
license: "free of charge since 2025-06-26, but NOT open data — access is per \
subscriber under SCB's terms of use, accepted when the credential is \
issued"
.into(),
// The register is organisations and workplaces. An enskild firma is a
// natural person trading under their own name, which is why this is not
// simply `Open` — but nothing lands until the layout is known, so the
// classification is the conservative one and gets revisited with the key.
gdpr: GdprClass::Open,
auth: AuthKind::ApiKey,
// `orgnr` would be the spine that joins Bolagsverket, GLEIF and FI — and it
// is NOT declared, because this collector emits no columns at all yet.
// Declaring the join key korp wants would be declaring a column that cannot
// land, which is precisely the O3 defect.
join_keys: vec![],
ts_pointer: None,
freshness: Some(crate::freshness::Freshness::AlwaysFull {
why: "nobody has read this API — the endpoint, the response shape and any \
incremental parameter all arrive with the credential, so any \
freshness claim here would be invented"
.into(),
}),
business_key: Some(crate::history::BusinessKey::append_only(
"the response layout is issued WITH the credential (SCB's postbeskrivning \
is chosen per subscriber), so no column is known to exist — an `orgnr` \
key declared now would name a column nothing emits",
)),
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: "".into(),
fields: Vec::new(),
}
}
/// Build a fully custom adapter (the config-driven escape hatch).
pub fn custom(name: &str, records_pointer: &str, fields: Vec<FieldMap>) -> Self {
Self {
name: name.into(),
description: format!("custom REST/JSON adapter `{name}`"),
license: "per-source".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec![],
ts_pointer: None,
freshness: None,
business_key: None,
next_pointer: None,
page_walk: None,
dedupe_pointer: None,
max_pages: DEFAULT_MAX_PAGES,
derived: Vec::new(),
last_read: std::sync::Mutex::new(None),
records_pointer: records_pointer.into(),
fields,
}
}
}
#[async_trait::async_trait]
impl Collector for RestCollector {
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
self.human_doc_card()
}
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: self.name.clone(),
source_kind: SourceKind::Rest,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["json".into()],
license: self.license.clone(),
gdpr: self.gdpr,
auth: self.auth,
join_keys: self.join_keys.clone(),
description: self.description.clone(),
}
}
/// **The REST presets' credentials, BY NAME.**
///
/// One `match` on the preset name, deliberately shaped like
/// [`source_target`](Collector::source_target) just below it: the same adapter
/// serves ~30 endpoints, and the fact that distinguishes them is which name they
/// read. Anything not listed keeps the derived declaration — for an
/// [`AuthKind::None`] preset that is "no secret", which is TRUE, and for a keyed
/// preset nobody has named yet it is an
/// [`undeclared`](crate::requires::SecretReq::undeclared) requirement, which reads
/// as `NeedsConfig` and never as runnable.
fn requirements(&self) -> crate::requires::Requirements {
use crate::requires::{Requirements, SecretReq};
let base = Requirements::open(self.source_target().source_endpoint);
match self.name.as_str() {
// OpenSky migrated to Keycloak OAuth2 client-credentials in 2025; the
// anonymous quota 401s. Two names, both required.
"opensky-flights" | "opensky-europe-bbox" => base.with_secrets(vec![
SecretReq::new(
"OPENSKY_CLIENT_ID",
"the OpenSky OAuth2 client id (Account ▸ API client, 2025 Keycloak flow)",
),
SecretReq::new(
"OPENSKY_CLIENT_SECRET",
"the OpenSky OAuth2 client secret",
),
]),
// `/tracks` is the older credential-gated arm (HTTP basic).
"opensky-tracks" => base.with_secrets(vec![
SecretReq::new("OPENSKY_USER", "an OpenSky account username (the /tracks arm)"),
SecretReq::new("OPENSKY_PASS", "the OpenSky account password"),
]),
"aisstream" => base.with_secrets(vec![SecretReq::new(
"AISSTREAM_API_KEY",
"a free AISStream.io API key",
)]),
"destatis-genesis" => base.with_secrets(vec![SecretReq::new(
"DESTATIS_GENESIS_TOKEN",
"a free Destatis GENESIS account token",
)]),
"mobility-database" | "mobility-database-rt" => base.with_secrets(vec![SecretReq::new(
"MOBILITY_DB_TOKEN",
"a free Mobility Database API refresh token",
)]),
// **Bolagsverket Värdefulla datamängder — declared, and NOT runnable here.**
//
// Neither variable is registered on this box, and that is the point: with
// both named, `Requirements::readiness` returns `NeedsConfig` and the ▶
// Collect refusal (and the daemon's) reads
// "cannot run — secret BOLAGSVERKET_CLIENT_ID is not set …;
// secret BOLAGSVERKET_CLIENT_SECRET is not set …", so the 📡 Collectors pane
// says it is WAITING FOR CREDENTIALS instead of pretending, or landing zero
// rows with no reason. The `identitetsbeteckning` field is required for the
// same reason: `POST /organisationer` takes ONE organisation per call, so a
// run with no orgnr configured cannot succeed and must not offer a live ▶.
"bolagsverket-organisationer" => base
.with_secrets(vec![
SecretReq::new(
BOLAGSVERKET_CLIENT_ID_ENV,
"the Bolagsverket API client id — free, self-service: register an \
application for `VärdefullaDatamängder` v1 at \
portal.api.bolagsverket.se (no agreement, no fee)",
),
SecretReq::new(
BOLAGSVERKET_CLIENT_SECRET_ENV,
"the matching client secret — korp exchanges the pair for a bearer \
token at the OAuth2 client-credentials endpoint; the secret itself \
is never stored in config",
),
])
.with_fields(vec![crate::requires::FieldReq::new(
"identitetsbeteckning",
"the organisationsnummer to look up — POST /organisationer resolves ONE \
identity per request, so there is no bulk read to fall back on",
)]),
// **SCB Företagsregistret — the lane's blocker, declared so it refuses BY
// NAME rather than by 401.**
//
// Two unmet requirements, and they are different KINDS of missing, which is
// why both have to be said:
//
// * the SOURCE is `Missing::PlaceholderSource` — SCB does not publish the
// base url anywhere; it is issued with the credentials, so the honest
// state is "the endpoint must be configured", exactly as for `rss-atom`;
// * the SECRET is `Missing::Secret { name: SCB_FORETAGSREGISTRET_KEY }`,
// with the mail address in the `why` so the refusal an operator reads
// also tells them how to unblock it.
//
// ONE secret and not three, on purpose. SCB's authorisation is a certificate
// plus password today and becomes an API key in September 2026 (their own
// words, and that is this month). Declaring all three would make the
// collector demand credentials from BOTH regimes at once and it could never
// be satisfied. Declaring the key alone matches what SCB says it will issue.
//
// If a CERTIFICATE arrives instead, this collector is still blocked and for
// a second reason: `crate::http` has no client-certificate path — no
// `Identity`, no PKCS#12 — so mutual TLS is a separate row and not a config
// change. That is stated on the card rather than discovered at the wire.
"scb-foretagsregistret" => base.with_secrets(vec![SecretReq::new(
SCB_FORETAGSREGISTRET_KEY_ENV,
"the SCB Företagsregistret API key — FREE since 2025-06-26 but not \
self-service: request it by mail to scbforetag@scb.se, naming the \
organisation, who accepts the terms of use, who receives the credential, \
and which data layout (postbeskrivning) you want. Without it korp's SCB \
half is PxWeb aggregates only — SNI × storleksklass × år, never an orgnr",
)]),
_ => base,
}
}
fn source_target(&self) -> SourceTarget {
// AIS lands a time-stamped position table the facett map ANIMATES; the
// other REST presets land plain warehouse tables.
if self.name == "digitraffic-ais" {
SourceTarget::map_scene(
SourceEndpoint::Ais,
DIGITRAFFIC_AIS_URL,
"ais_positions",
"osm3d-flight",
)
} else if self.name == "opensky-flights" {
SourceTarget::map_scene(
SourceEndpoint::GenericRest,
"https://opensky-network.org/api/states/all",
"opensky_flights",
"osm3d-flight",
)
} else if self.name == "opensky-europe-bbox" {
SourceTarget::map_scene(
SourceEndpoint::GenericRest,
"https://opensky-network.org/api/states/all?lamin=34.5&lomin=-25.0&lamax=72.0&lomax=45.0",
"opensky_europe_bbox",
"osm3d-flight",
)
} else if self.name == "opensky-tracks" {
SourceTarget::map_scene(
SourceEndpoint::OpenSkyTracks,
"https://opensky-network.org/api/tracks/all?icao24=&time=0",
"opensky_tracks",
"osm3d-flight",
)
} else if self.name == "aisstream" {
SourceTarget::map_scene(
SourceEndpoint::Ais,
"https://stream.aisstream.io/v0/stream",
"aisstream_positions",
"osm3d-flight",
)
} else if self.name == "gbfs-citybikes" {
SourceTarget::map_scene(
SourceEndpoint::GenericRest,
"https://gbfs.urbansharing.com/oslobysykkel.no/free_bike_status.json",
"gbfs_citybikes",
"osm3d-flight",
)
} else if self.name == "osm-overpass" {
// Point features land in `osm_features`, overlaid by the `osm3d-poi`
// scene (rendered OVER the osm_ways basemap).
SourceTarget::map_scene(
SourceEndpoint::Osm,
"https://overpass-api.de/api/interpreter",
"osm_features",
"osm3d-poi",
)
} else if self.name == "smhi-weather" {
SourceTarget::map_scene(
SourceEndpoint::GenericRest,
"https://opendata-download-metobs.smhi.se/api/version/latest/parameter/1.json",
"smhi_weather_stations",
"osm3d",
)
} else if self.name == "smhi-hydroobs" {
SourceTarget::warehouse(
SourceEndpoint::GenericRest,
"https://opendata-download-hydroobs.smhi.se/api/version/latest/parameter/1/station-set/all/period/latest-hour/data.json",
"smhi_hydroobs",
)
} else if self.name == "metno-forecast" {
SourceTarget::map_scene(
SourceEndpoint::GenericRest,
"https://api.met.no/weatherapi/locationforecast/2.0/compact",
"metno_forecast",
"osm3d",
)
} else if self.name == "gbif-occurrences" {
// Biodiversity occurrence points → the `osm3d-poi` overlay scene.
SourceTarget::map_scene(
SourceEndpoint::GenericRest,
"https://api.gbif.org/v1/occurrence/search?hasCoordinate=true&limit=20",
"gbif_occurrences",
"osm3d-poi",
)
} else if self.name == "worldbank-indicators" {
SourceTarget::warehouse(
SourceEndpoint::GenericRest,
"https://api.worldbank.org/v2/country/all/indicator/SP.POP.TOTL?format=json&per_page=100",
"worldbank_indicators",
)
} else if self.name == "restcountries" {
SourceTarget::warehouse(
SourceEndpoint::GenericRest,
"https://restcountries.com/v3.1/all?fields=name,cca2,cca3,region,subregion,population,latlng",
"restcountries",
)
} else if self.name == "mobility-database" {
// The feed CATALOGUE lands in a plain warehouse table (a meta-index
// that downstream GTFS/GTFS-RT/GBFS collects join back on by feed_id).
SourceTarget::warehouse(
SourceEndpoint::MobilityDatabase,
"https://api.mobilitydatabase.org/v1/gtfs_feeds",
"mobility_database_feeds",
)
} else if self.name == "mobility-database-rt" {
// The GTFS-Realtime feed CATALOGUE — a meta-index whose rows drive the
// gtfs-rt-vehicles protobuf reader (join by feed_id / latest_url).
SourceTarget::warehouse(
SourceEndpoint::MobilityDatabaseRt,
"https://api.mobilitydatabase.org/v1/gtfs_rt_feeds",
"mobility_database_rt_feeds",
)
} else if self.name == "bolagsverket-organisationer" {
// Company registry → a plain warehouse table (the orgnr join spine).
//
// Repointed 2026-09-03 off the DEAD `api.bolagsverket.se/naringslivsregistret/v1`
// URL onto the FREE Värdefulla datamängder API — see the preset's own doc
// comment for the measurements. One writer for the URL: the constant.
SourceTarget::warehouse(
SourceEndpoint::Bolagsverket,
BOLAGSVERKET_ORGANISATIONER_URL,
"bolagsverket_organisationer",
)
} else if self.name == "scb-foretagsregistret" {
// A PLACEHOLDER, deliberately — SCB issues the endpoint with the
// credentials and publishes it nowhere. See `SCB_FORETAGSREGISTRET_URL`.
SourceTarget::warehouse(
SourceEndpoint::ScbForetagsregistret,
SCB_FORETAGSREGISTRET_URL,
"scb_foretagsregistret",
)
} else if self.name == "domstol-rattspraxis" {
// Court decisions → a plain warehouse table. One writer for the url: the
// constant, because the sort parameter in it is load-bearing.
SourceTarget::warehouse(
SourceEndpoint::DomstolRattspraxis,
DOMSTOL_RATTSPRAXIS_URL,
"domstol_rattspraxis",
)
} else if self.name == "eea-wise-water" {
SourceTarget::warehouse(
SourceEndpoint::EeaWise,
"https://discodata.eea.europa.eu/sql?query=SELECT%20TOP%20100%20*%20FROM%20%5BWISE_SOE%5D.%5Bv2r1%5D.%5BWaterbase_S_WISE6_AggregatedData%5D",
"eea_wise_water",
)
} else if self.name == "destatis-genesis" {
SourceTarget::warehouse(
SourceEndpoint::DestatisGenesis,
"https://www-genesis.destatis.de/genesisWS/rest/2020/catalogue/tables",
"destatis_genesis_tables",
)
} else if self.name == "cbs-odata" {
SourceTarget::warehouse(
SourceEndpoint::CbsOData,
"https://opendata.cbs.nl/ODataApi/odata/37296eng/TypedDataSet",
"cbs_odata",
)
} else if self.name == "riksdagen-dokument" {
// The parliament's document register → a plain warehouse table. The URL is
// the preset's own constant rather than a literal here, because `sz`, the
// sort and the doktyp filter are all load-bearing (see the preset's doc
// comment) and must have exactly one writer.
SourceTarget::warehouse(
SourceEndpoint::RiksdagenDokument,
RIKSDAGEN_DOKUMENTLISTA_URL,
"riksdagen_dokument",
)
} else {
let endpoint = match self.name.as_str() {
"polisen-events" => "https://polisen.se/api/events",
// Kolada v2 is GONE. Measured live 2026-09-03: `GET
// api.kolada.se/v2/kpi?per_page=2` → **HTTP 410**, 65 B, body
// `{"error": "This endpoint is deprecated. Please use /v3 instead."}`;
// v3 → **200**, 955 B, same `{"values":[…]}` envelope; v1 → 404. Only
// the URL moved, so the preset keeps its pointer and its join keys.
"kolada-kpi" => "https://api.kolada.se/v3/kpi",
// The FULL query, filter and page size included: the filter name is
// the thing the lane note got wrong (`entity.registeredAt.id` → 400),
// so it belongs in the one place a run actually reads rather than in
// prose. Measured 2026-09-04: 119 764 SE records, 200 per page.
"gleif-lei" => GLEIF_SWEDEN_URL,
"smhi-metobs" => "https://opendata-download-metobs.smhi.se/api",
_ => "",
};
SourceTarget::warehouse(SourceEndpoint::GenericRest, endpoint, self.name.clone())
}
}
/// **The graph these rows honestly make** (lane EC O7).
///
/// Two of the 24 REST presets carry a relationship worth drawing; the rest declare
/// `None`, which is not an omission — a state vector, a country list or a KPI
/// catalogue row names no second entity, and an edge invented for it would be
/// korp's opinion rather than the source's data.
fn graph_mapping(&self) -> Option<crate::graph_mapping::GraphMapping> {
use crate::graph_mapping::{GraphMapping, SpineLookup};
match self.name.as_str() {
"polisen-events" => Some(
GraphMapping::new(
"polisen_events",
"Polisen — händelser",
"polisen_events",
"polisen's API carries no kommunkod — only `location_name` free text. \
A place edge exists only where that text matched a SCB name exactly; \
the report names every value that did not.",
)
.with_spine("location_name", SpineLookup::ByName)
.vertex(
"PolisEvent",
"id",
&["datetime", "name", "type", "summary", "url", "location_name", "gps"],
)
// The event's own `type` column IS the type node's identity.
.edge("OF_TYPE", ("PolisEvent", "id", "id"), ("EventType", "type", "type"))
// NULL where the name matched no kommun ⇒ the sink drops the edge.
.edge(
"IN_MUNICIPALITY",
("PolisEvent", "id", "id"),
("Municipality", "municipality_code", "kommun_code"),
)
.edge(
"IN_COUNTY",
("PolisEvent", "id", "id"),
("County", "county_code", "county_code"),
),
),
// **The JOIN BRIDGE as a graph** (lane EC O4). This is the mapping that makes
// an orgnr and a LEI one entity: FI's insider register names an issuer by
// LEI only, TED names a supplier by name and country, and Bolagsverket
// names a company by orgnr. `Company` is keyed on `orgnr` — the SAME
// property Bolagsverket's own rows key it on — so the two collectors MERGE
// onto one node instead of building two.
"gleif-lei" => Some(
GraphMapping::new(
"gleif_lei",
"GLEIF — LEI ↔ organisationsnummer",
"gleif_lei",
"GLEIF carries no kommunkod, so nothing here joins to the kommun \
spine; and `registeredAs` is self-declared by the entity at \
registration, so a company that mistyped its own orgnr bridges to \
nothing — the edge is dropped rather than pointed at a NULL, and \
the row still lands with its `registered_as` verbatim as the \
evidence.",
)
.vertex(
"Lei",
"lei",
&[
"legal_name",
"entity_status",
"legal_form",
"city",
"postal_code",
"registration_status",
"last_update",
"managing_lou",
],
)
// `orgnr` is the DERIVED digits-only column, never `registered_as`:
// Bolagsverket keys `Company` on the un-hyphenated orgnr, and an edge
// to `559586-2557` would MERGE a second Company node for the same
// firm — the graph would silently double and every count be wrong.
.edge("HAS_LEI", ("Company", "orgnr", "orgnr"), ("Lei", "lei", "lei")),
),
// **A court register as a graph** (lane EC O6). Two entities the source
// itself names, and one relationship that is a fact of the data rather than
// korp's opinion: a publication was issued BY a court, and the court's code
// is the primary key of the source's own `GET /api/v1/domstolar` codelist.
//
// The publication GROUP is the second edge and the interesting one: a
// decision and its referat share a `gruppKorrelationsnummer` (607 distinct
// over 610 publications, 2026-09-05), which is how "these two documents are
// the same case" is expressed — a fact no join on case number can recover,
// since case numbers repeat.
"domstol-rattspraxis" => Some(
GraphMapping::new(
"domstol_rattspraxis",
"Domstolsverket — rättspraxis",
"domstol_rattspraxis",
"rättsområde, nyckelord and lagrum are MULTI-VALUED at the source \
(5 of 610 publications name two legal areas) and land as one joined \
string, so none of them is an edge here: an edge from a joined cell \
would MERGE a node named `Miljömål; Skadestånd` that no court ever \
used. There is also no orgnr anywhere — a company is named only in \
free text inside `sammanfattning`, so nothing joins to the \
Bolagsverket/GLEIF company spine and no such edge is invented.",
)
.vertex(
"CourtPublication",
"id",
&[
"typ",
"publiceringsform",
"avgorandedatum",
"publiceringstid",
"mal_nummer",
"referat_nummer",
"benamning",
"sammanfattning",
"innehall_tecken",
"bilaga_fillagring_id",
],
)
// `Court` is keyed on the code, never the name: `domstolNamn` is display
// text ("Svea hovrätts hyresrättsliga avgöranden" is a *view* of Svea
// hovrätt), while `domstolKod` is the codelist key the API itself
// publishes at `/api/v1/domstolar`.
.vertex("Court", "domstol_kod", &["domstol_namn"])
.edge(
"DECIDED_BY",
("CourtPublication", "id", "id"),
("Court", "domstol_kod", "domstol_kod"),
)
// Same case, more than one document. The group node is keyed on the
// correlation number the source assigns; the publication points AT it.
.edge(
"IN_CASE_GROUP",
("CourtPublication", "id", "id"),
(
"CaseGroup",
"grupp_korrelationsnummer",
"grupp_korrelationsnummer",
),
),
),
// **A parliamentary document names its issuer, and that is a real edge.**
//
// `organ` is the department that laid the proposition —
// `Justitiedepartementet` 54, `Finansdepartementet` 40,
// `Socialdepartementet` 24 of page 1's 200 (measured 2026-09-05). That is the
// source's own statement about a second entity, not korp's opinion, so it is
// drawn — but only 371 of 400 rows over two live pages carry it, and the
// limitation says so rather than letting an operator discover it as a count
// that does not add up.
"riksdagen-dokument" => Some(
GraphMapping::new(
"riksdagen_dokument",
"Riksdagen — dokument och avsändare",
"riksdagen_dokument",
"`organ` is present on 371 of 400 documents (92.8 %, measured over a \
live two-page collect 2026-09-05), so ~7 % of RiksdagDokument nodes \
get NO `ISSUED_BY` edge — the endpoint is NULL and the sink drops \
it, which is correct and is why the count of documents will not \
equal the count of edges. And where it IS present it is a NAME, not \
a code, in a mixed namespace: a department spelled out for a \
proposition (`Justitiedepartementet`), a utskott abbreviation for a \
betänkande (`MJU`). Those MERGE onto `Organisation` by the same \
`publisher` property DCAT and SMHI use, so a department joins the \
rest of the graph only where the two sources spell it identically — \
nothing here normalises them. There is also no place: a document is \
not located anywhere, so no column resolves against the kommun/län \
spine.",
)
.vertex(
"RiksdagDokument",
"dok_id",
&[
"titel",
"doktyp",
"subtyp",
"rm",
"beteckning",
"nummer",
"datum",
"publicerad",
"status",
"dokument_url_html",
],
)
// The document's own `doktyp` IS the type node's identity — the same
// shape polisen's `PolisEvent -> EventType` uses.
.edge(
"OF_TYPE",
("RiksdagDokument", "dok_id", "dok_id"),
("DokumentTyp", "doktyp", "doktyp"),
)
// `Organisation` is DCAT's label and SMHI's, and all three must key it
// on `publisher` — `graph_mapping::key_conflicts` is what keeps them
// agreeing, and a second id property here would MERGE a second node per
// department and silently double every count.
.edge(
"ISSUED_BY",
("RiksdagDokument", "dok_id", "dok_id"),
("Organisation", "organ", "publisher"),
),
),
"smhi-weather" => Some(
GraphMapping::new(
"smhi_stations",
"SMHI — mätstationer",
"smhi_weather",
"SMHI's station list carries lat/lon but no kommunkod, and a point is \
not a municipality until something resolves it — korp does not, so \
there is no station→kommun edge.",
)
.vertex(
"WeatherStation",
"station_key",
&["name", "lat", "lon", "height", "active"],
)
// `Organisation` is DCAT's label too, and both must key it on
// `publisher` — `graph_mapping::key_conflicts` is what keeps them agreeing.
.edge(
"OWNED_BY",
("WeatherStation", "station_key", "station_key"),
("Organisation", "owner", "publisher"),
),
),
_ => None,
}
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(self.rest_schema())
}
fn incremental_key(&self) -> Option<&str> {
self.ts_pointer.as_deref()
}
fn freshness(&self) -> crate::freshness::Freshness {
self.freshness
.clone()
.unwrap_or_else(|| crate::freshness::Freshness::derived(self.incremental_key()))
}
fn business_key(&self) -> crate::history::BusinessKey {
self.business_key.clone().unwrap_or_else(|| {
// The preset's own field map IS the emitted column set, so the derivation
// is checked against it rather than hoped over. See
// `BusinessKey::derived_over`.
// The field map PLUS the derived columns — both land, so both count as
// emitted. `join_keys` for a bridge collector names the derived key
// (`orgnr`), and intersecting against the field map alone would have made
// every such collector read as append-only for the wrong reason.
let emitted: Vec<String> = self
.fields
.iter()
.map(|f| f.column.clone())
.chain(self.derived.iter().map(|d| d.column.clone()))
.collect();
crate::history::BusinessKey::derived_over(&self.join_keys, Some(&emitted))
})
}
async fn collect(
&self,
cfg: &CollectorCfg,
bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
use crate::freshness::{Freshness, Plan};
// The preset's own published rate cap (Digitraffic: 60/min ⇒ 1/s), not a
// one-size-fits-all 10/s.
let http = cfg.http(self.max_per_sec());
// The collector's OWN auth unless the caller overrode `cfg.auth` — the same
// precedence gribnc uses. Without it, korp's fallback config (built for any
// collector absent from `known_sources()`) would send `AuthConfig::None` at an
// OAuth2 source and collect a 401.
let auth_cfg = match &cfg.auth {
AuthConfig::None => self.default_auth(),
other => other.clone(),
};
// resolve_async so an OAuth2 client-credentials source (OpenSky's 2025
// Keycloak flow, Bolagsverket's WSO2 one) can fetch its bearer token; every
// other auth delegates to the sync resolver. A missing credential errors HERE,
// naming the env var.
let auth = auth_cfg.resolve_async().await?;
// …then the constant identification headers the source mandates (Digitraffic).
let headers = self.required_headers();
let auth: Box<dyn super::http::Auth> = if headers.is_empty() {
auth
} else {
Box::new(super::http::WithHeaders::new(auth, headers))
};
let fresh = self.freshness();
let plan = bounds.plan.clone().unwrap_or_else(|| Plan::of(&fresh, None));
// ── the URL this run actually asks for ───────────────────────────────
//
// A server-side lower bound (polisen's `?DateTime=`) is what makes a re-run
// cheaper in BYTES and not only in rows. The parameter carries the mark
// truncated to the server's own granularity — asking a day-granular parameter
// for a full timestamp returns nothing at all.
let url = match &plan {
Plan::Window {
param: Some(p),
param_value: Some(v),
..
} => {
let sep = if cfg.base_url.contains('?') { '&' } else { '?' };
format!(
"{}{sep}{}={}",
cfg.base_url,
p,
super::http::percent_encode(v)
)
}
Plan::Offset {
param,
offset: Some(o),
} => {
let sep = if cfg.base_url.contains('?') { '&' } else { '?' };
format!(
"{}{sep}{param}={}",
cfg.base_url,
super::http::percent_encode(o)
)
}
_ => cfg.base_url.clone(),
};
// ── the fetch ────────────────────────────────────────────────────────
let (if_none_match, if_modified_since) = match &plan {
Plan::Conditional {
if_none_match,
if_modified_since,
} => (if_none_match.clone(), if_modified_since.clone()),
_ => (None, None),
};
// A POST source (Bolagsverket's `/organisationer`, 405 on GET) sends its
// declared body; everything else keeps the conditional GET, validators and all.
// A POST has no `If-None-Match` to send, so it reports no validators and the
// freshness plan degrades to a full read rather than lying about a 304.
let fetched = match self.request_body(cfg)? {
Some(body) => {
let text = http
.post_text(&url, body, "application/json", auth.as_ref())
.await?;
super::http::Fetched::Body {
bytes: text.into_bytes(),
etag: None,
last_modified: None,
}
}
None => {
http.get_conditional(
&url,
auth.as_ref(),
if_none_match.as_deref(),
if_modified_since.as_deref(),
)
.await?
}
};
let transferred = fetched.transferred();
let (etag, last_modified) = fetched.validators();
// ── 304: the source PROVED it did not change ─────────────────────────
//
// Nothing is re-landed and nothing is red. The checkpoint keeps the previous
// mark (the runner re-stamps it) and the reason reads back through
// `CollectStats::ragnarok` as an explained empty.
let Some(body) = fetched.text() else {
*self.last_read.lock().expect("checkpoint mutex") =
Some(Checkpoint::default().unchanged(transferred));
return Ok(CollectStats {
table: cfg.table.clone(),
reason: Some(crate::freshness::unchanged_reason(
&self.name,
"304 Not Modified",
)),
..Default::default()
});
};
let mut json: Value = serde_json::from_str(&body)?;
let mut records_ptr: String = self.records_pointer.clone();
let mut transferred = transferred;
// ── the CURSOR CRAWL ─────────────────────────────────────────────────
//
// A source that answers 200 records and a `links.next` has not answered the
// question; it has answered the first two hundredths of it. So the pages are
// followed and their record arrays concatenated into ONE array, and the
// records pointer becomes `""` — every path below is then unchanged, which is
// what keeps the delta filter, the watermark and the field map from each
// needing a paginated twin.
let mut pages = 1usize;
let mut cap_hit = false;
let mut deduped = 0usize;
if self.next_pointer.is_some() || self.page_walk.is_some() {
let first = records(&json, &records_ptr)?.clone();
let mut last_len = first.len();
let mut all: Vec<Value> = first;
// The urls already asked for. A `next` that points at a page already read
// is a SOURCE BUG, and it must not become an infinite loop inside korp:
// the crawl stops and lands what it has.
let mut seen: std::collections::BTreeSet<String> = [url.clone()].into();
let mut next = self.next_page(&json, &url, pages, last_len);
while let Some(u) = next {
if pages >= self.max_pages {
cap_hit = true;
break;
}
if !seen.insert(u.clone()) {
break;
}
// No validators on a page beyond the first: a `304` on page 7 of a
// crawl would mean "this page is unchanged", not "the source is", and
// treating it as the latter would silently truncate the table.
let f = http.get_conditional(&u, auth.as_ref(), None, None).await?;
transferred += f.transferred();
let Some(b) = f.text() else { break };
let v: Value = serde_json::from_str(&b)?;
let recs = records(&v, &records_ptr)?;
last_len = recs.len();
all.extend(recs.iter().cloned());
pages += 1;
next = self.next_page(&v, &url, pages, last_len);
}
// **The same record, read twice, is ONE row.** A source whose paging is not
// a stable total order repeats records across page boundaries; landing both
// copies doubles them in any sink that appends. A record with no value at
// the pointer is KEPT — an absent key cannot prove a duplicate, and dropping
// it would turn "we do not know" into "we have seen this".
if let Some(ptr) = &self.dedupe_pointer {
let before = all.len();
let mut keys: std::collections::BTreeSet<String> = Default::default();
all.retain(|r| match r.pointer(ptr).and_then(|v| v.as_str()) {
Some(k) => keys.insert(k.to_string()),
None => true,
});
deduped = before - all.len();
}
json = Value::Array(all);
records_ptr = String::new();
}
let records_ptr = records_ptr;
// ── the row-level delta ──────────────────────────────────────────────
//
// When the adapter declares a datetime field, NARROW the records to `bounds`
// under the declared ORDER (lexical is wrong for polisen's single-digit hour)
// and record the high-watermark. Without a declared field there is no
// incremental key → a full snapshot, and `freshness()` says so.
let order = match &fresh {
Freshness::QueryWindow { order, .. } => *order,
Freshness::FileSeries { order, .. } => *order,
_ => crate::freshness::MarkOrder::Lexical,
};
let mut watermark: Option<String> = None;
let rb = match &self.ts_pointer {
Some(tp) => {
let recs = records(&json, &records_ptr)?;
let kept = filter_by_bounds_ordered(recs, tp, &bounds, order);
// The high-watermark is taken over the WHOLE response, not over the
// rows kept: a window that returned only records we already had must
// still advance past them, or the next run asks for the same slice
// forever.
let all: Vec<&Value> = recs.iter().collect();
watermark = max_mark(&all, tp, order)
.or_else(|| bounds.since.clone());
build_batch(&kept, &self.fields)?
}
None => parse_records(&json, &records_ptr, &self.fields)?,
};
// The normalisations a JOIN KEY needs, applied to the batch the field map
// built — so the derived column is a column of the landed table and not
// something every downstream query has to remember to write.
let rb = apply_derived(rb, &self.derived)?;
let mut ck = Checkpoint::default()
.with_validators(etag, last_modified)
.with_bytes(transferred);
if let Some(m) = watermark {
ck.mark = m;
}
let n = rb.num_rows() as u64;
if n == 0 {
// A window that returned nothing NEW is an unchanged source, not a
// failure — the same first-class explained zero a 304 produces.
let reason = if bounds.since.is_some() {
ck.unchanged = true;
crate::freshness::unchanged_reason(
&self.name,
&format!(
"no record after {}",
bounds.since.as_deref().unwrap_or("the watermark")
),
)
} else if bounds.until.is_some() {
format!("{} returned no records in bounds", self.name)
} else {
format!("{} returned no records", self.name)
};
*self.last_read.lock().expect("checkpoint mutex") = Some(ck);
return Ok(CollectStats {
table: cfg.table.clone(),
reason: Some(reason),
..Default::default()
});
}
*self.last_read.lock().expect("checkpoint mutex") = Some(ck);
sink.push(rb)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
// A crawl that stopped at its ceiling landed rows AND is incomplete, and
// both have to be said: a truncated table that reports plain success is a
// table an operator believes is the whole register. A dedupe is the same
// class of fact — it is EVIDENCE the source's paging is not a stable total
// order, and the records it repeated are records it may equally have
// skipped — so it is reported beside it rather than swallowed.
reason: crate::rest::join_reasons(&[
cap_hit.then(|| {
format!(
"{} stopped at the {}-page ceiling and the source HAS more — this \
table is INCOMPLETE. Raise `with_max_pages` or narrow the query.",
self.name, self.max_pages
)
}),
(deduped > 0).then(|| {
format!(
"{} dropped {deduped} record(s) the crawl read TWICE across page \
boundaries — the source's paging is not a stable total order, so \
it may equally have SKIPPED records nothing here can see.",
self.name
)
}),
]),
})
}
/// CDC checkpoint: what the last [`collect`] read — the newest timestamp it saw,
/// the validators the source served, the bytes it cost, and whether the source
/// proved unchanged. `None` until a collect has run.
///
/// The collector REPORTS; it does not persist. The runner decides whether to
/// advance the stored mark, and does so only after the sink has flushed.
async fn checkpoint(&self) -> Result<Option<Checkpoint>> {
Ok(self.last_read.lock().expect("checkpoint mutex").clone())
}
}
#[cfg(test)]
mod tests {
use super::*;
/// OpenSky `/states` POSITIONAL-array records map to a flight position table —
/// the offline DATA contract (no network). Emits a `nornir-testmatrix` cell
/// (via the crate `functional_status` wrapper — reuses edda's testmatrix, a
/// no-op without `--features testmatrix`) so `nornir test` SEES the collector.
#[test]
fn opensky_states_parse_into_a_flight_position_table() {
// Two aircraft states in OpenSky's positional-array shape.
let json: Value = serde_json::json!({
"time": 1_700_000_000i64,
"states": [
["4b1815", "SWR123 ", "Switzerland", 1_700_000_000i64, 1_700_000_000i64,
8.55, 47.45, 10972.8, false, 231.0, 87.3, 0.0, null, 11277.6, "1000", false, 0],
["3c6750", "DLH456 ", "Germany", 1_700_000_000i64, 1_700_000_000i64,
11.78, 48.35, 3200.0, false, 180.0, 270.0, 5.2, null, 3300.0, "2000", false, 0]
]
});
let c = RestCollector::opensky();
let rb = parse_records(&json, c.records_pointer(), c.fields()).expect("parse states");
assert_eq!(rb.num_rows(), 2, "two aircraft states parsed");
let schema = rb.schema();
let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
for want in ["icao24", "callsign", "lat", "lon", "true_track", "velocity"] {
assert!(
The file goes on; the console shows all of it.