src/lib.rs
//! **korp-collectors** — the standalone LEAF of korp's open-data collector layer.
//!
//! This crate carries the pluggable import of Swedish + EU open data into Apache
//! Iceberg (through **skade**), built **per access-METHOD** not per agency. It was
//! extracted out of `korp::collect` so that it depends on **neither korp nor
//! facett** — which breaks the facett⇄korp dependency cycle: `facett-demo` and
//! `korp` can both depend on this leaf, while `facett` itself depends on nothing
//! of korp. (Mirrors how `osm-katana` and `Skidbladnir` were extracted.)
//!
//! The shape matches korp's `.nornir/collectors.md`: one object-safe [`Collector`]
//! trait (one shape, both ingest paths behind it), a [`CollectorMeta`] discovery
//! record so the UI/CLI/test-matrix enumerate + license/GDPR-gate every one, an
//! Arrow [`BatchSink`](sink::BatchSink) that lands `RecordBatch`es, and a
//! path-agnostic [`run`] runner (behind `skade-sink`) that wires
//! `skade::Warehouse::table_or_create` → `Table::append` per the korp-lite
//! (Spark-free, DEFAULT) profile.
//!
//! ## Features
//! * `parsers` (default) — the trait/registry surface + shared HTTP core + every
//! LIGHT pure-Rust parser (DCAT, PxWeb, ENTSO-E, REST, Trafikverket) + the pure
//! OSM admin tag-filter/trie. No heavy geo dep, no Iceberg engine.
//! * `osm` — the OSM geometry connectors ([`osm`]) via `osm-katana`, reading a LOCAL
//! OSM container. Regions are cut from the torrented planet, never fetched per-region.
//! * `skade-sink` — the [`BatchSink::iceberg`](sink::BatchSink::iceberg) Iceberg
//! backend + the [`run`] runner.
//! * `testmatrix` — the tests' `functional_status` self-reports become real
//! nornir test-matrix rows (off by default; no nornir dep otherwise).
mod direction;
pub use direction::Direction;
/// **Build stamp** — the crate version + the git commit it was built from.
///
/// The commit hash is captured at build time by the pure-filesystem `build.rs`
/// (zero shell / zero `git`, per the LAW) into `COLLECTORS_GIT_HASH`, falling back
/// to `unknown` when no `.git` is reachable. Consumers surface it through
/// [`version_string`] on `--version`/`-V` and on the daemon health surface
/// ([`daemon::BiosBus::version`]).
pub mod build {
/// The crate version (`CARGO_PKG_VERSION`), e.g. `0.1.7`.
pub const VERSION: &str = env!("CARGO_PKG_VERSION");
/// The crate name (`CARGO_PKG_NAME`) — `korp-collectors`.
pub const NAME: &str = env!("CARGO_PKG_NAME");
/// The git commit this build was stamped from, or `unknown`.
pub const GIT_HASH: &str = env!("COLLECTORS_GIT_HASH");
}
/// The canonical version line — `<name> v<version> (<git-hash>)`, e.g.
/// `korp-collectors v0.1.7 (38e6b2c…)`. Printed by a daemon runner on
/// `--version`/`-V` (before any daemon boot) and posted onto the BIOS health bus.
pub fn version_string() -> String {
format!("{} v{} ({})", build::NAME, build::VERSION, build::GIT_HASH)
}
/// Handle a `--version` / `-V` flag for a daemon runner: if present in `args`,
/// print [`version_string`] to stdout and return `true` (the caller MUST exit
/// **before** booting any daemon). Returns `false` when no version flag is present.
///
/// ```no_run
/// if korp_collectors::handle_version_flag(std::env::args().skip(1)) {
/// std::process::exit(0);
/// }
/// // … boot the daemon …
/// ```
pub fn handle_version_flag<I, S>(args: I) -> bool
where
I: IntoIterator<Item = S>,
S: AsRef<str>,
{
let hit = args
.into_iter()
.any(|a| matches!(a.as_ref(), "--version" | "-V"));
if hit {
println!("{}", version_string());
}
hit
}
/// **What a collector NEEDS before it can run** — the declaration (source URL, named
/// secrets, required config fields) and the four-state [`Readiness`](requires::Readiness)
/// derived from it. Unconditional: it is pure `serde`, and both the one-shot [`Collector`]
/// ingest registry and the [`daemon`] registry answer with it, so a host has ONE way to ask
/// "can this run, and if not, what is missing, by name?".
pub mod provenance;
pub mod roster;
pub mod requires;
pub mod human_doc;
/// **How a source CHANGES** — the sibling declaration to [`requires`]: the six
/// [`Freshness`](freshness::Freshness) kinds (conditional fetch · file series · query
/// window · append-only log · byte-range columnar · honestly-always-full), the
/// [`ReadMark`](freshness::ReadMark) one successful run leaves behind, the
/// [`MarkStore`](freshness::MarkStore) seam a host persists it through, and the
/// [`Plan`](freshness::Plan) the next run derives from the two. Unconditional and pure
/// `serde` for the same reason `requires` is.
pub mod freshness;
/// **How landed rows are HISTORIZED** — the business key a collector declares, the
/// system columns that turn an append-only Iceberg table into one row per key plus a
/// readable history, and the honest per-backend matrix of what merge/upsert each
/// warehouse can actually do. See the module docs for why the shape is
/// append-with-validity rather than `MERGE INTO`.
pub mod history;
/// **How a collector's landed table becomes typed nodes and real edges** — the
/// DECLARATION, sibling of [`history`] on the graph side. Lane EC O7: this knowledge
/// used to live in korp as a hand-written `static SOURCES: [SourceMapping; 5]` carrying
/// Rust function pointers, so a collector could not describe its own graph and the
/// pipeline roster was a list that could drift from the registry.
pub mod graph_mapping;
/// The collector **DAEMON** architecture (feature `daemon`) — the object-safe
/// `Config + State + Collector` trait trio, the `CollectorCtx` seams (S3 bucket,
/// jera phase handle, log, BIOS bus), and the self-registering daemon registry.
/// See `.nornir/collectors-daemon-design.md`. Distinct from the one-shot
/// [`Collector`] ingest trait above.
#[cfg(feature = "daemon")]
pub mod daemon;
/// The **`Runtime` substrate trait** (feature `daemon`) — the interface that decides
/// WHERE a [`daemon`] collector (or a knut pipeline) runs (Jera-metal / Jera-distributed
/// / Kube), decoupled from its logic. Interface only — the concrete `JeraRuntime` /
/// `KubeRuntime` live in korp. See `.nornir/runtime-aortas-design-2026-07-29.md`.
#[cfg(feature = "daemon")]
pub mod runtime;
/// **The systemd substrate** (feature `systemd`) — a collector as a unit of its own,
/// installed, started, stopped and uninstalled **through Skidbladnir**. Pure adapter:
/// [`runtime::UnitSpec`] → `skidbladnir::product::ComponentSpec`, no unit text rendered
/// here, no purge owned here.
#[cfg(feature = "systemd")]
pub mod runtime_systemd;
/// **The collector STATE STORE** (feature `daemon`) — the redb-backed ledger where each
/// step of the a–e chain records that it FINISHED, written by the producer at the instant
/// its own API says done.
///
/// It replaces asking the filesystem. "Is this planet complete?" used to be answered by
/// the file's NAME, and then by `st_blocks * 512 >= st_size` — an inference about a state
/// librqbit already knew, over a proxy that sparse files, compressing filesystems,
/// preallocation and reflink copies each break independently. See [`state`] for the
/// contract and its red probes.
#[cfg(feature = "daemon")]
pub mod state;
/// **The LIVE REGISTER** (feature `daemon`) — where the phase AND the facet of every
/// collector a process is nursing are written down, so **another process can read them**.
///
/// Live state used to be readable only from inside the supervising process. While that
/// held, a service outside it could not see the collectors it was supposed to nurse, which
/// is the difference between a supervisor and a thread pool. Same discipline as [`state`]:
/// the producer knows, the producer writes it down, the reader reads the register — and
/// every row carries WHO wrote it, because live state read from a file has one failure
/// mode a read from memory cannot have: the writer can be gone.
#[cfg(feature = "daemon")]
pub mod register;
/// **The `Acquire` seam** (feature `daemon`) — for artefacts too big to hold in memory.
///
/// [`daemon::Fetcher`] returns `Vec<u8>` and is right for the twenty collectors whose
/// bodies are megabytes. This is the second seam, for the ones whose bodies are not: it
/// streams to a durable destination, resumes an interrupted transfer, reports progress
/// where the supervisor can see it, honours a stop, and checks a per-format completeness
/// declaration before the artefact ever wears its final name. **The discriminator is
/// size, not transport.**
#[cfg(feature = "daemon")]
pub mod acquire;
/// **The NURSERY** (feature `service`) — the standalone collector service: one process
/// that supervises MANY collectors, restarts them with backoff, publishes each one's live
/// state to the [`register`] so anything outside can read it, and waits for the network
/// through Skidbladnir before starting work.
///
/// This is what a foreign project installs. It carries no OSM: the geometry stack is
/// behind `osm-daemon`, and a service built without it has a registry of pure-parser
/// collectors and none of the planet.
#[cfg(feature = "service")]
pub mod nursery;
/// The `osm-cut` daemon (feature `osm-daemon`) — the reference implementor of the
/// [`daemon`] architecture: planet → region catalog → tiles → ONLINE.
#[cfg(feature = "osm-daemon")]
pub mod osm_cut;
/// **The `osm-cut` cut POST-CONDITIONS and SAFETY PROPERTIES** (feature
/// `osm-daemon`) — the checks that FAIL a run which produced a wrong pyramid, and the
/// never-destroy-the-only-copy / rollback / disk-headroom / stamp-last properties that
/// used to be a human being careful. See [`osm_cut_verify`].
#[cfg(feature = "osm-daemon")]
pub mod osm_cut_verify;
/// **WAVE 1 — the Iceberg-sink collector daemons** (feature `daemon`): `dcat`,
/// `pxweb`, `osm-admin` adopted to the [`daemon`] architecture (design §11). Unlike
/// [`osm_cut`] (an S3-sink), they fetch an external source, parse it with the
/// crate's existing pure parsers, and land Arrow batches through the
/// [`daemon::WarehouseSink`] seam. They prove the pattern is generic over both sink
/// kinds; the remaining ~17 collectors are the mechanical follow-on.
#[cfg(feature = "daemon")]
pub mod iceberg_daemons;
#[cfg(feature = "parsers")]
mod surface {
use anyhow::Result;
use arrow_schema::Schema as ArrowSchema;
use crate::Direction;
use crate::http;
use crate::sink::BatchSink;
/// Which ingest path a collector lands through. Per the two collector classes
/// in `collectors.md`: **A** Rust-native (zero-JVM, DEFAULT) vs **B** Spark
/// Connect. Every collector here is `RustNative`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum CollectorPath {
/// Path A — parse in Rust → Arrow → skade (in-process, no JVM).
RustNative,
/// Path B — drive Spark Connect for JVM-only sources (Kafka/JDBC/DataSourceV2).
Spark,
}
/// The access-method family a collector belongs to (§4 of `sweden-open-data.md`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum SourceKind {
/// DCAT-AP / RDF catalogue crawl (discovery layer).
Dcat,
/// PxWeb / JSON-stat statistical cubes.
PxWeb,
/// SDMX-JSON statistical messages (Eurostat, ECB — the series/observation
/// dimension split). Distinct from PxWeb's JSON-stat2 cube shape.
Sdmx,
/// OGC WFS / WMS / OGC-API-Features geodata. (roadmap)
Ogc,
/// GTFS / GTFS-RT transit. (roadmap)
Gtfs,
/// Plain REST / JSON (incl. ENTSO-E XML).
Rest,
/// Bulk file download (coarse legacy bucket). **Superseded** by the finer
/// file-format variants below (`FlatFile`/`Spreadsheet`/…); kept so the
/// existing OSM collectors and any serialized meta stay valid. New file
/// collectors pick the specific variant.
BulkFile,
// ── finer file-format families (additive split of `BulkFile`) ──────────
// These let the registry/test-matrix/UI group the bulk-file catalog by
// shape instead of one coarse bucket. Serde-additive: existing variants
// (incl. `BulkFile`) keep their names, so no persisted meta breaks.
/// Delimited / fixed-width / mainframe flat files (CSV, TSV, fixed-width,
/// EBCDIC+COBOL-copybook).
FlatFile,
/// Spreadsheets (xlsx / xls / ods).
Spreadsheet,
/// Statistical-package dumps (SAS / SPSS / Stata / DBF).
StatDump,
/// Relational dumps (pg_dump / mysqldump / SQLite files).
SqlDump,
/// Already-columnar analytical files (Parquet / ORC / Avro / Arrow IPC).
Columnar,
/// Geospatial bulk files (Shapefile / GeoJSON / GeoPackage / FlatGeobuf).
Geo,
/// Business/message interchange (X12 / EDIFACT / HL7 / ISO 20022 / SIE).
Edi,
}
/// GDPR / sensitivity class — the segmentation line `collectors.md` enforces.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum GdprClass {
/// Open data, no personal data.
Open,
/// Carries personal data (personnummer etc.) — legal-basis gated.
Personal,
/// Licensed / agreement-gated (Nord Pool, Lantmäteriet account).
Gated,
}
/// What credential a source needs — drives the pluggable [`http::Auth`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum AuthKind {
/// No auth (most SE open data).
None,
/// Self-serve API key (Trafiklab, Trafikverket-key, …).
ApiKey,
/// Free token (ENTSO-E).
Token,
/// Signed agreement / account (Trafikverket agreement, Lantmäteriet).
Agreement,
/// Paid license (Nord Pool) — default-OFF.
Licensed,
}
impl AuthKind {
/// The stable word for this auth kind — what a requirement declaration quotes
/// when a collector needs credentials but has not named them.
pub fn as_str(self) -> &'static str {
match self {
AuthKind::None => "None",
AuthKind::ApiKey => "ApiKey",
AuthKind::Token => "Token",
AuthKind::Agreement => "Agreement",
AuthKind::Licensed => "Licensed",
}
}
}
/// Discovery record: name, DIRECTION, source kind, formats, capabilities,
/// PATH, LICENSE + GDPR class — so UI/CLI/test-matrix enumerate & gate every
/// collector.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct CollectorMeta {
pub name: String,
/// Data-flow direction — every [`Collector`] is a [`Direction::Source`]
/// (it PULLS data in); the [`BatchSink`] is the [`Direction::Sink`]. An
/// orthogonal axis to `source_kind` (the transport/format family).
pub direction: Direction,
pub source_kind: SourceKind,
pub path: CollectorPath,
pub formats: Vec<String>,
pub license: String,
pub gdpr: GdprClass,
pub auth: AuthKind,
/// Join keys this collector carries (orgnummer, kommunkod, elområde, …) —
/// the entity-resolution spine.
pub join_keys: Vec<String>,
pub description: String,
}
/// Per-collect configuration. `base_url` + `params` are config-driven so one
/// connector-per-method is reused across agencies; `auth` is pluggable and
/// reads secrets from env (never hardcoded).
#[derive(Debug, Clone, Default)]
pub struct CollectorCfg {
/// API base URL (per-source — same client, different base URL).
pub base_url: String,
/// Target Iceberg table name.
pub table: String,
/// Per-collector parameters (table id / selection / domain codes / pointers).
pub params: serde_json::Value,
/// How to authenticate this source (resolves from env at collect time).
pub auth: AuthConfig,
/// **A canned response table, when this run must not touch the network.**
/// `None` in every shipped path. Set by a hermetic test (and by a daemon tick
/// driven offline) so [`Collector::collect`] reads fixtures — see
/// [`http::Canned`]. It rides on the config rather than on 22 per-collector
/// constructors, which is what makes ONE seam reach every collector.
pub canned: Option<http::Canned>,
}
impl CollectorCfg {
pub fn new(base_url: impl Into<String>, table: impl Into<String>) -> Self {
Self {
base_url: base_url.into(),
table: table.into(),
params: serde_json::Value::Null,
auth: AuthConfig::None,
canned: None,
}
}
/// Serve this collect from `canned` instead of the network.
pub fn with_canned(mut self, canned: http::Canned) -> Self {
self.canned = Some(canned);
self
}
/// **The ONE constructor a `collect()` builds its HTTP client with.**
///
/// `max_per_sec` is the source's own published rate cap, which is the reason
/// the collector — not korp — owns this number. Routing every collector
/// through here is what carries [`canned`](CollectorCfg::canned) into all of
/// them: a `HttpCore::new` written in a leaf is a twin that silently ignores
/// the fixture and goes live.
pub fn http(&self, max_per_sec: u32) -> http::HttpCore {
http::HttpCore::new(max_per_sec).with_canned(self.canned.clone())
}
pub fn with_params(mut self, params: serde_json::Value) -> Self {
self.params = params;
self
}
pub fn with_auth(mut self, auth: AuthConfig) -> Self {
self.auth = auth;
self
}
}
/// Pluggable auth, resolved from environment variables at collect time.
/// Secrets are NEVER stored in config — only the env-var name is, so offline
/// tests and committed configs carry no credentials.
#[derive(Debug, Clone, Default)]
pub enum AuthConfig {
#[default]
None,
/// Add `?{param}={env(VAR)}` to the query (ENTSO-E securityToken, Trafiklab key).
QueryKey { param: String, env: String },
/// Add header `{header}: {env(VAR)}` (Trafikverket-style key).
HeaderKey { header: String, env: String },
/// `Authorization: Bearer {env(VAR)}`.
Bearer { env: String },
/// **OAuth2 client-credentials** — fetch a bearer token from `token_url`
/// using the client id/secret in the named env vars, then send
/// `Authorization: Bearer {token}`. OpenSky's live API migrated to this
/// Keycloak flow in 2025 (anonymous / basic-auth now 401s for the
/// authenticated quota). Resolved via the async [`resolve_async`] (the
/// token fetch is a network round-trip); secrets stay in env.
///
/// [`resolve_async`]: AuthConfig::resolve_async
OAuth2ClientCredentials {
token_url: String,
id_env: String,
secret_env: String,
},
}
impl AuthConfig {
/// Resolve to a live [`http::Auth`] by reading the named env var. Returns
/// an error (with the missing var name) when a credentialled source has no
/// secret in the environment — fixtures never hit this path.
///
/// [`OAuth2ClientCredentials`](AuthConfig::OAuth2ClientCredentials) needs a
/// network token fetch, so it is NOT resolvable here — use
/// [`resolve_async`](AuthConfig::resolve_async) for it (this returns an
/// error naming that requirement so a mis-wired sync call fails loudly).
pub fn resolve(&self) -> Result<Box<dyn http::Auth>> {
Ok(match self {
AuthConfig::None => Box::new(http::NoAuth),
AuthConfig::QueryKey { param, env } => {
Box::new(http::QueryKey::new(param, env_var(env)?))
}
AuthConfig::HeaderKey { header, env } => {
Box::new(http::HeaderKey::new(header, env_var(env)?))
}
AuthConfig::Bearer { env } => Box::new(http::Bearer::new(env_var(env)?)),
AuthConfig::OAuth2ClientCredentials { .. } => {
return Err(anyhow::anyhow!(
"OAuth2 client-credentials auth needs an async token fetch — use \
AuthConfig::resolve_async()"
));
}
})
}
/// Async resolution: [`OAuth2ClientCredentials`](AuthConfig::OAuth2ClientCredentials)
/// performs its client-credentials token fetch here (reading the id/secret
/// env vars, erroring with the missing var name if absent); every other
/// variant delegates to the sync [`resolve`](AuthConfig::resolve). Collectors
/// call this so any source — OpenSky included — can carry an OAuth2 token.
pub async fn resolve_async(&self) -> Result<Box<dyn http::Auth>> {
match self {
AuthConfig::OAuth2ClientCredentials {
token_url,
id_env,
secret_env,
} => {
let id = env_var(id_env)?;
let secret = env_var(secret_env)?;
let token =
http::oauth2_client_credentials_token(token_url, &id, &secret).await?;
Ok(Box::new(http::Bearer::new(token)))
}
other => other.resolve(),
}
}
}
fn env_var(name: &str) -> Result<String> {
std::env::var(name)
.map_err(|_| anyhow::anyhow!("missing credential env var `{name}` (auth-gated source)"))
}
/// Incremental marks (CDC seam): which slice to collect.
///
/// `since`/`until` were declared with the crate and, until the freshness work,
/// were set in exactly two places — both of them tests. [`plan`](Bounds::plan)
/// is what fills them on a real run: the runner resolves the collector's
/// [`Freshness`](crate::freshness::Freshness) against the stored
/// [`ReadMark`](crate::freshness::ReadMark) and hands the answer down here.
#[derive(Debug, Clone, Default)]
pub struct Bounds {
pub since: Option<String>,
pub until: Option<String>,
/// **What this run should fetch** — the resolved incremental plan (conditional
/// validators, the last file consumed, the log offset, the version token), or
/// `None` for the unconditional full read every caller got before.
///
/// A collector reads this to build its request; it can NOT write a mark
/// through it. Only the runner, holding a landed row count, advances a mark —
/// see [`crate::freshness`].
pub plan: Option<crate::freshness::Plan>,
}
impl Bounds {
/// The bounds a [`Plan`](crate::freshness::Plan) implies — `since` from the
/// plan's watermark, the plan itself carried whole for the kinds that need
/// more than a timestamp.
pub fn plan(plan: crate::freshness::Plan) -> Self {
Bounds {
since: plan.since(),
until: None,
plan: Some(plan),
}
}
/// Is this run a delta at all? `false` for a full read (including every run
/// made before a plan was resolved).
pub fn is_delta(&self) -> bool {
self.plan.as_ref().is_some_and(|p| p.is_delta())
}
}
/// Where a CDC-aware collector got to (reuse of the knut-bifrost TableState seam).
///
/// The collector reports it; the RUNNER decides whether to persist it, and does so
/// only after the sink has flushed. See [`crate::freshness`] for why that ordering
/// is the whole safety property.
#[derive(Debug, Clone, Default)]
pub struct Checkpoint {
/// The resume value in the freshness kind's own terms — a watermark, a
/// filename, an offset, a version token. Empty when the validators ARE the
/// mark (a pure conditional fetch).
pub mark: String,
/// The `ETag` the source served this run, for the next `If-None-Match`.
pub etag: Option<String>,
/// The `Last-Modified` the source served, for the next `If-Modified-Since`.
pub last_modified: Option<String>,
/// Bytes actually fetched this run — `0` on a `304`, and that zero is what
/// makes "unchanged" measurable rather than asserted.
pub bytes: u64,
/// The source proved unchanged this run (a `304`, no new file, no record after
/// the watermark). The runner re-stamps the previous mark instead of replacing
/// it, and RAGNARÖK reads the 0-row collect as an explained empty.
pub unchanged: bool,
}
impl Checkpoint {
/// A checkpoint carrying only a resume mark — what a watermark collector reports.
pub fn at(mark: impl Into<String>) -> Self {
Checkpoint {
mark: mark.into(),
..Default::default()
}
}
/// Attach the validators the next conditional request should send back.
pub fn with_validators(
mut self,
etag: Option<String>,
last_modified: Option<String>,
) -> Self {
self.etag = etag;
self.last_modified = last_modified;
self
}
/// Attach the byte count this run actually transferred.
pub fn with_bytes(mut self, bytes: u64) -> Self {
self.bytes = bytes;
self
}
/// Mark this run as having found the source unchanged.
pub fn unchanged(mut self, bytes: u64) -> Self {
self.unchanged = true;
self.bytes = bytes;
self
}
/// The [`Delta`](crate::freshness::Delta) this checkpoint reports for a run
/// that landed `rows` rows — the bridge from what the COLLECTOR saw to what
/// the RUNNER persists.
pub fn delta(&self, rows: u64, how: &str) -> crate::freshness::Delta {
if self.unchanged {
return crate::freshness::Delta::Unchanged {
how: how.to_string(),
bytes: self.bytes,
};
}
crate::freshness::Delta::Landed {
rows,
bytes: self.bytes,
mark: self.mark.clone(),
etag: self.etag.clone(),
last_modified: self.last_modified.clone(),
}
}
}
/// Result of one collect run. RAGNARÖK: `rows == 0` is RED unless `reason`
/// explains why (provably empty source / license-gated / parse gap).
#[derive(Debug, Clone, Default)]
pub struct CollectStats {
pub rows: u64,
pub batches: u64,
pub table: String,
/// Why a 0-row (or partial) collect happened — non-`None` makes an empty
/// collect honest rather than silently RED.
pub reason: Option<String>,
}
/// The sentinel [`CollectStats::reason`] the [`run`] runner stamps on a
/// 0-row collect the COLLECTOR left unexplained.
///
/// `run()` has always annotated an unexplained empty with this string so an
/// operator reading the stats sees something. The catch: once written, the
/// stats became indistinguishable from a collector that HAD explained its
/// emptiness — an unexplained empty laundered itself into an honest one, i.e.
/// exactly the *silently green* case RAGNARÖK forbids. Naming the sentinel
/// makes it recognisable again: [`CollectStats::ragnarok`] treats it as
/// **not an explanation**, so a collector that forgets its `reason` stays RED
/// through the runner.
pub const RAGNAROK_UNEXPLAINED: &str =
"collect landed 0 rows (RAGNARÖK: source empty or parse gap?)";
/// The RAGNARÖK verdict of one collect run — derived from
/// [`CollectStats`], never stored, so it costs no field and breaks no
/// downstream struct literal.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Ragnarok {
/// Rows landed. Green.
Rows,
/// Zero rows, and the collector SAID WHY (provably empty window,
/// license-gated, known parse gap). Honest — not a failure.
ExplainedEmpty,
/// Zero rows and NOBODY explained it (`reason` absent, or only the
/// runner's [`RAGNAROK_UNEXPLAINED`] placeholder). **RED.**
Red,
}
impl Ragnarok {
/// Stable lowercase tag for JSON / matrix rows.
pub fn as_str(self) -> &'static str {
match self {
Ragnarok::Rows => "rows",
Ragnarok::ExplainedEmpty => "explained-empty",
Ragnarok::Red => "red",
}
}
}
impl CollectStats {
/// The RAGNARÖK verdict for this run. Rows → [`Ragnarok::Rows`]; zero
/// rows with a collector-supplied `reason` → [`Ragnarok::ExplainedEmpty`];
/// zero rows with no reason (or only the runner's
/// [`RAGNAROK_UNEXPLAINED`] placeholder) → [`Ragnarok::Red`].
pub fn ragnarok(&self) -> Ragnarok {
if self.rows > 0 {
return Ragnarok::Rows;
}
match self.reason.as_deref() {
None => Ragnarok::Red,
Some(r) if r == RAGNAROK_UNEXPLAINED => Ragnarok::Red,
Some(_) => Ragnarok::ExplainedEmpty,
}
}
/// Is this run RED (an unexplained empty)? The one-line gate a caller —
/// a CLI, the Ingest UI, a scheduler — checks instead of eyeballing
/// `rows`/`reason`.
pub fn is_red(&self) -> bool {
self.ragnarok() == Ragnarok::Red
}
/// [`is_red`](Self::is_red) as a `Result`: `Err` naming the table when the
/// run is an unexplained empty, `Ok(self)` otherwise. Lets a caller write
/// `collector.collect(..).await?.ragnarok_check()?`.
pub fn ragnarok_check(&self) -> Result<&Self> {
if self.is_red() {
return Err(anyhow::anyhow!(
"RAGNARÖK: collect into `{}` landed 0 rows and no reason was given \
(empty is RED, never silently green)",
if self.table.is_empty() {
"<unnamed table>"
} else {
&self.table
}
));
}
Ok(self)
}
}
/// The concrete open-data ENDPOINT family a collector pulls FROM — the
/// `source` half of the [`SourceTarget`] descriptor. Finer than
/// [`SourceKind`] (the transport/format): ENTSO-E, Trafikverket and AIS all
/// share `SourceKind::Rest` but are distinct endpoints the test-matrix wants
/// to iterate one row each. This is what makes every collector an
/// **enumerable, visible** `(collector, source, target)` test candidate.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum SourceEndpoint {
/// DCAT-AP catalogue crawl (dataportal.se RDF / EU SPARQL).
DcatCatalog,
/// PxWeb / JSON-stat2 statistical API (SCB, Energimyndigheten, …).
PxWebApi,
/// Eurostat SDMX-JSON 2.1 statistics API (EU-wide cubes: geo × time × measure).
Eurostat,
/// ENTSO-E Transparency Platform (EU electricity).
EntsoE,
/// Trafikverket real-time road/rail XML-query API.
Trafikverket,
/// AIS ship-position feed.
Ais,
/// OpenStreetMap PBF/XML geometry extract (basemap ways).
Osm,
/// OpenStreetMap `boundary=administrative` relations (borders).
OsmAdmin,
/// Generic REST / JSON endpoint (Polisen, Kolada, SMHI, …).
GenericRest,
/// Bulk flat file (CSV / fixed-width / EBCDIC copybook).
FlatFile,
/// OGC WFS (Web Feature Service) — GML / GeoJSON feature collections
/// (Lantmäteriet, SGU, Naturvårdsverket geodata).
Wfs,
/// GTFS static transit feed (a zip of `stops`/`routes`/`trips` CSV).
Gtfs,
/// GTFS-Realtime `VehiclePositions` feed (a protobuf `FeedMessage` — live
/// transit vehicle positions; distinct from the static GTFS zip).
GtfsRt,
/// Bolagsverket company registry (Näringslivsregistret — orgnummer spine).
Bolagsverket,
/// Nord Pool day-ahead electricity spot prices (per bidding zone).
NordPool,
/// USGS real-time earthquake GeoJSON feed (global seismicity points).
Usgs,
/// RSS 2.0 / Atom 1.0 syndication feed (alert/news/press item streams).
Feed,
/// OpenAQ open air-quality measurements API (global pollutant readings).
OpenAq,
/// Energy-Charts (Fraunhofer ISE) public-power generation-mix API
/// (Europe-wide electricity by production type).
EnergyCharts,
/// Mobility Database (`api.mobilitydatabase.org`, CC0) — the global
/// catalogue of GTFS / GTFS-RT / GBFS feed records (the successor to the
/// retired TransitFeeds). A **meta-collector**: its rows are feed
/// descriptors that auto-drive the crate's GTFS / GTFS-RT collectors.
MobilityDatabase,
/// ERDDAP tabledap / griddap scientific-data server (EMODnet + global
/// ocean/atmosphere ERDDAP deployments) — uniform `.csv` / `.json`
/// gridded/tabular access over any dataset id.
Erddap,
/// Copernicus Climate Data Store (CDS) retrieve API
/// (`cds.climate.copernicus.eu/api/retrieve/v1/`) — ERA5 / reanalysis
/// GRIB & NetCDF retrievals (key from env `CDSAPI_KEY`).
CopernicusCds,
/// Deutscher Wetterdienst (DWD) ICON open-data NWP — GRIB2 forecast
/// fields published on `opendata.dwd.de` (no auth).
DwdIcon,
/// EEA WISE / Discodata — the European Environment Agency's water-quality
/// (WISE-SoE Waterbase) data over the Discodata SQL REST API
/// (`discodata.eea.europa.eu/sql`), returning a `results` row array.
EeaWise,
/// Destatis GENESIS-Online — the German Federal Statistical Office national
/// statistics REST API (`www-genesis.destatis.de/genesisWS/rest/2020/`);
/// the table-catalogue `List` of statistical-table descriptors.
DestatisGenesis,
/// CBS Open Data — Statistics Netherlands' national statistics OData API
/// (`opendata.cbs.nl/ODataApi/odata/<table>/TypedDataSet`), returning a
/// `value` row array of typed statistical observations.
CbsOData,
/// INSEE BDM (Banque de Données Macro-économiques) — the French national
/// statistics **SDMX 2.1** web service (`bdm.insee.fr/series/sdmx/data`);
/// rides the shared SDMX-JSON flattener (zero new parsing).
Insee,
/// ONS Beta — the UK Office for National Statistics **SDMX** API
/// (`api.beta.ons.gov.uk`); rides the shared SDMX-JSON flattener.
Ons,
/// ISTAT — the Italian national statistics **SDMX 2.1** web service
/// (`esploradati.istat.it/SDMXWS/rest/data`); rides the shared SDMX-JSON
/// flattener.
Istat,
/// World Bank **SDMX** REST data API (`api.worldbank.org/v2/sdmx/rest/data`)
/// — the SDMX face of the World Bank indicators (distinct from the plain
/// REST `worldbank-indicators`); rides the shared SDMX-JSON flattener.
WorldBankSdmx,
/// Copernicus Climate Data Store **ERA5** reanalysis — the fixed CDS
/// `retrieve/v1` endpoint (`cds.climate.copernicus.eu/api/retrieve/v1/`)
/// feeding the GRIB/NetCDF **value** decode (`gribnc` mode=values). Key
/// from env `CDSAPI_KEY`.
CopernicusEra5,
/// Copernicus Atmosphere Data Store (ADS) **CAMS** air-quality — the fixed
/// ADS `retrieve/v1` endpoint (`ads.atmosphere.copernicus.eu/api/retrieve/v1/`)
/// feeding the GRIB/NetCDF value decode. Key from env `ADSAPI_KEY`.
CopernicusCams,
/// Mobility Database **GTFS-Realtime** feed catalogue
/// (`api.mobilitydatabase.org/v1/gtfs_rt_feeds`, CC0) — the RT-feed face of
/// the Mobility Database meta-collector; its rows are GTFS-RT feed
/// descriptors that auto-drive the [`GtfsRt`](SourceEndpoint::GtfsRt) reader.
MobilityDatabaseRt,
/// OpenSky Network **flight TRACK** — the recorded trajectory of one
/// aircraft (`opensky-network.org/api/tracks/all`): an ordered `path[]` of
/// `(time, lat, lon, baro_altitude, heading, on_ground)` waypoints. The
/// "flight recorder" (time-ordered) face, distinct from the instantaneous
/// `/states` snapshot behind [`GenericRest`](SourceEndpoint::GenericRest).
OpenSkyTracks,
/// ENTSO-E Transparency **day-ahead PRICES** (`documentType=A44`) — the
/// energy-market spot-price face of [`EntsoE`](SourceEndpoint::EntsoE),
/// carrying `price.amount` per zone × hour (the free-token open route to
/// day-ahead prices; the licensed twin is [`NordPool`](SourceEndpoint::NordPool)).
EntsoeDayAhead,
/// Domstolsverket **rättspraxis** — the PUH publication API
/// (`rattspraxis.etjanst.domstol.se/api/v1/publiceringar`, keyless): every
/// published judgment, decision, referat and notis from Sweden's higher
/// courts. Page-NUMBER paginated with no cursor of any kind, which is why
/// [`rest::PageWalk`](crate::rest::PageWalk) exists.
DomstolRattspraxis,
/// SCB **Företagsregistret** — the per-organisationsnummer business register
/// REST service. NOT the keyless PxWeb Statistical Database
/// ([`PxWebApi`](SourceEndpoint::PxWebApi)): this one is credential-gated and
/// its base URL is issued with the credentials rather than published, which is
/// why the collector behind it refuses BY NAME instead of guessing a host.
ScbForetagsregistret,
/// **Riksdagen öppna data** — the Swedish parliament's document register
/// (`data.riksdagen.se/dokumentlista`, keyless): propositioner, betänkanden,
/// motioner and SFS, 670 547 documents behind one JSON URL (measured
/// 2026-09-05). Distinct from [`Feed`](SourceEndpoint::Feed), which is
/// regeringen.se's syndication of the GOVERNMENT's side of the same
/// legislative pipeline: a proposition appears in both, as an RSS item there
/// and as a `dok_id` row here.
RiksdagenDokument,
}
impl SourceEndpoint {
/// Stable lowercase tag for JSON / matrix rows.
pub fn as_str(self) -> &'static str {
match self {
SourceEndpoint::DcatCatalog => "dcat-catalog",
SourceEndpoint::PxWebApi => "pxweb-api",
SourceEndpoint::Eurostat => "eurostat",
SourceEndpoint::EntsoE => "entsoe",
SourceEndpoint::Trafikverket => "trafikverket",
SourceEndpoint::Ais => "ais",
SourceEndpoint::Osm => "osm",
SourceEndpoint::OsmAdmin => "osm-admin",
SourceEndpoint::GenericRest => "generic-rest",
SourceEndpoint::FlatFile => "flat-file",
SourceEndpoint::Wfs => "wfs",
SourceEndpoint::Gtfs => "gtfs",
SourceEndpoint::GtfsRt => "gtfs-rt",
SourceEndpoint::Bolagsverket => "bolagsverket",
SourceEndpoint::NordPool => "nordpool",
SourceEndpoint::Usgs => "usgs",
SourceEndpoint::Feed => "feed",
SourceEndpoint::OpenAq => "openaq",
SourceEndpoint::EnergyCharts => "energy-charts",
SourceEndpoint::MobilityDatabase => "mobility-database",
SourceEndpoint::Erddap => "erddap",
SourceEndpoint::CopernicusCds => "copernicus-cds",
SourceEndpoint::DwdIcon => "dwd-icon",
SourceEndpoint::EeaWise => "eea-wise",
SourceEndpoint::DestatisGenesis => "destatis-genesis",
SourceEndpoint::CbsOData => "cbs-odata",
SourceEndpoint::Insee => "insee",
SourceEndpoint::Ons => "ons",
SourceEndpoint::Istat => "istat",
SourceEndpoint::WorldBankSdmx => "worldbank-sdmx",
SourceEndpoint::CopernicusEra5 => "copernicus-era5",
SourceEndpoint::CopernicusCams => "copernicus-cams",
SourceEndpoint::MobilityDatabaseRt => "mobility-database-rt",
SourceEndpoint::OpenSkyTracks => "opensky-tracks",
SourceEndpoint::EntsoeDayAhead => "entsoe-day-ahead",
SourceEndpoint::DomstolRattspraxis => "domstol-rattspraxis",
SourceEndpoint::ScbForetagsregistret => "scb-foretagsregistret",
SourceEndpoint::RiksdagenDokument => "riksdagen-dokument",
}
}
}
/// Where a collector's parsed data LANDS — the `target` half of the
/// [`SourceTarget`] descriptor.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum TargetKind {
/// A skade / Iceberg **warehouse table** (the default landing zone).
WarehouseTable,
/// A warehouse table that a **facett map scene** renders (basemap ways,
/// admin borders, or an animated position overlay). Same Iceberg table —
/// this variant records that a map consumer exists downstream.
MapScene,
/// In-memory **Arrow** `RecordBatch`es only (the parser/fixture contract —
/// no warehouse in the loop).
Arrow,
}
impl TargetKind {
/// Stable lowercase tag for JSON / matrix rows.
pub fn as_str(self) -> &'static str {
match self {
TargetKind::WarehouseTable => "warehouse-table",
TargetKind::MapScene => "map-scene",
TargetKind::Arrow => "arrow",
}
}
}
/// The typed **source → target descriptor** every [`Collector`] declares via
/// [`Collector::source_target`]. It names the open-data endpoint a collector
/// pulls FROM (`source` + a representative `source_endpoint` URL) and where
/// the parsed rows land (`target` + the representative `target_table`, plus
/// the `map_scene` a facett map consumes when the target feeds one:
/// clickable-Sweden / osm3d / osm3d-flight). Enumerating these over
/// [`registry_descriptors`] gives the test-matrix its `(collector, source,
/// target)` tuples.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct SourceTarget {
/// The open-data endpoint family pulled FROM.
pub source: SourceEndpoint,
/// A representative endpoint (URL host / API root) — documentation + the
/// live sample fetch (`sample::*_try_get_real_sample_data`) use it.
pub source_endpoint: String,
/// Where the parsed data lands.
pub target: TargetKind,
/// The representative Iceberg/warehouse table the collector writes.
pub target_table: String,
/// The facett map scene this feeds, when any (`clickable-sweden` /
/// `osm3d` / `osm3d-flight`); `None` for pure warehouse targets.
pub map_scene: Option<String>,
}
impl SourceTarget {
/// A warehouse-table target (no map consumer).
pub fn warehouse(
source: SourceEndpoint,
endpoint: impl Into<String>,
table: impl Into<String>,
) -> Self {
Self {
source,
source_endpoint: endpoint.into(),
target: TargetKind::WarehouseTable,
target_table: table.into(),
map_scene: None,
}
}
/// A map-scene target — a warehouse table a named facett scene renders.
pub fn map_scene(
source: SourceEndpoint,
endpoint: impl Into<String>,
table: impl Into<String>,
scene: impl Into<String>,
) -> Self {
Self {
source,
source_endpoint: endpoint.into(),
target: TargetKind::MapScene,
target_table: table.into(),
map_scene: Some(scene.into()),
}
}
/// One-line `source → target` label for matrix rows / UI.
pub fn label(&self) -> String {
match &self.map_scene {
Some(scene) => format!(
"{} → {} ({}, scene={scene})",
self.source.as_str(),
self.target.as_str(),
self.target_table
),
None => format!(
"{} → {} ({})",
self.source.as_str(),
self.target.as_str(),
self.target_table
),
}
}
}
/// The one collector shape — both ingest paths sit behind it. Modeled on
/// `znippy_common::ArchiveTypePlugin` (discovery via `meta()`),
/// `knut_bifrost::TableSource` (async fetch + CDC state), `facett_core::Facet`.
#[async_trait::async_trait]
pub trait Collector: Send + Sync {
/// Discovery: name, direction, source kind, formats, PATH, LICENSE + GDPR.
fn meta(&self) -> CollectorMeta;
/// The typed **source → target descriptor**: which open-data endpoint this
/// collector pulls FROM and where the parsed rows land (warehouse table /
/// Arrow / map scene). Every collector DECLARES its own; the default
/// derives a warehouse target from [`CollectorMeta::source_kind`] so an
/// out-of-tree `Collector` still compiles, but the built-in ones override
/// it with a precise declaration so [`registry_descriptors`] enumerates a
/// real `(collector, source, target)` row for each.
fn source_target(&self) -> SourceTarget {
let m = self.meta();
let source = match m.source_kind {
SourceKind::Dcat => SourceEndpoint::DcatCatalog,
SourceKind::PxWeb => SourceEndpoint::PxWebApi,
SourceKind::Sdmx => SourceEndpoint::Eurostat,
SourceKind::Rest | SourceKind::Ogc | SourceKind::Gtfs => {
SourceEndpoint::GenericRest
}
_ => SourceEndpoint::FlatFile,
};
SourceTarget::warehouse(source, "", m.name)
}
/// **What this collector needs before it can run** — its source URL, its
/// secrets BY NAME and its required config fields
/// ([`crate::requires::Requirements`]).
///
/// The default derives the whole declaration from what the collector already
/// publishes — [`source_target`](Collector::source_target)'s endpoint and
/// [`meta`](Collector::meta)'s [`AuthKind`] — so no collector is silently
/// undeclared and an out-of-tree one still compiles. An auth kind cannot NAME a
/// variable, so anything but [`AuthKind::None`] derives
/// [`SecretReq::undeclared`](crate::requires::SecretReq::undeclared), which
/// reads as `NeedsConfig` and never as runnable. Override it wherever the
/// collector knows the real variable names.
///
/// This is the declaration a UI needs to tell RUNNABLE from UNRUNNABLE. Without
/// it every row got a ▶ Run button and `opensky-flights` answered
/// `korp-rpc-21: GET → builder error` — an empty endpoint discovered at the
/// wire instead of at the declaration.
fn requirements(&self) -> crate::requires::Requirements {
let m = self.meta();
crate::requires::Requirements::derived(
&self.source_target().source_endpoint,
m.auth.as_str(),
matches!(m.auth, AuthKind::None),
)
}
/// The Arrow schema this collector emits (drives `table_or_create`).
fn schema(&self, cfg: &CollectorCfg) -> Result<ArrowSchema>;
/// Stream parsed `RecordBatch`es to the sink. `bounds` carries incremental marks.
async fn collect(
&self,
cfg: &CollectorCfg,
bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats>;
/// **How does this source CHANGE?** — the sibling declaration to
/// [`requirements`](Collector::requirements), and the one fact a generic
/// runner cannot guess.
///
/// The default derives it from what the collector already publishes: a
/// collector with a declared per-record incremental key
/// ([`incremental_key`](Collector::incremental_key)) gets a local record
/// watermark — rows really are deltaed, and
/// [`saves_bytes`](crate::freshness::Freshness::saves_bytes) reports honestly
/// that the transfer is not. A collector with no key gets
/// [`AlwaysFull`](crate::freshness::Freshness::AlwaysFull), which SAYS it
/// re-reads everything rather than implying a delta that is not happening.
///
/// Override it wherever the collector knows how its source really changes —
/// exactly like `requirements()`.
fn freshness(&self) -> crate::freshness::Freshness {
crate::freshness::Freshness::derived(self.incremental_key())
}
/// The per-record field (RFC-6901 pointer) this collector's rows are dated by,
/// when it has one. The single fact
/// [`freshness`](Collector::freshness)' default is derived from; `None` means
/// the collector has no incremental key and re-reads its whole source.
fn incremental_key(&self) -> Option<&str> {
None
}
/// **What identifies a ROW across runs** — the business key, and the sibling
/// of [`freshness`](Collector::freshness) on the landing side.
///
/// The default derives it from
/// [`CollectorMeta::join_keys`](CollectorMeta::join_keys), which every
/// collector already publishes as "the entity-resolution spine" — **and then
/// checks that derivation against the collector's OWN schema**, because a spine
/// is written in the vocabulary of the domain (`coordinate`, `elområde`,
/// `REF_AREA`) and the landed rows carry columns (`lat`/`lon`, `in_domain`,
/// `geo`). Measured 2026-09-04: **29 of 53** collectors derived a key naming a
/// column they do not emit. A key that names nothing matches nothing, so the
/// merge it was wired into (lane EC O1) silently degrades to an append.
///
/// A key that does not survive that check becomes
/// [`append_only`](crate::history::BusinessKey::append_only) **whole**, never
/// the surviving subset — see
/// [`derived_over`](crate::history::BusinessKey::derived_over) for why the
/// subset is the dangerous answer.
fn business_key(&self) -> crate::history::BusinessKey {
let emitted = self.schema(&CollectorCfg::default()).ok().map(|s| {
s.fields()
.iter()
.map(|f| f.name().clone())
.collect::<Vec<String>>()
});
crate::history::BusinessKey::derived_over(&self.meta().join_keys, emitted.as_deref())
}
/// **The business key for THIS config** — the cfg-aware sibling, and the one a
/// runner should ask.
///
/// Most collectors emit a fixed column set and answer the same thing either
/// way, which is why the default simply delegates. But a **cube** collector
/// (SDMX, PxWeb/JSON-stat2) has a message-driven schema: its columns are one
/// per DIMENSION of the cube it was pointed at, plus `value`. Its identity is
/// therefore "every dimension column", which is knowable from the cfg and not
/// knowable without it — and a fixed guess is the worst answer available, since
/// a cube with an extra dimension (age, sex, SNI) would collapse every one of
/// its values into a single row.
///
/// Same shape as [`schema`](Collector::schema), for the same reason.
fn business_key_for(&self, _cfg: &CollectorCfg) -> crate::history::BusinessKey {
self.business_key()
}
/// **How this collector's landed table becomes typed nodes and real edges** —
/// the graph-side sibling of [`business_key`](Collector::business_key), and the
/// declaration a host derives a PIPELINE from.
///
/// `None` is a real answer and the common one: a table whose rows declare no
/// relationship to anything (an AIS position — mmsi, lat, lon, sog: a position
/// is not a place) must say so rather than be handed invented edges. It is what
/// korp's `NO_GRAPH` ledger said in prose, said by the collector instead.
///
/// The collector is the thing that knows what its columns mean, so this is
/// where the knowledge belongs. Until lane EC O7 it lived in korp as a
/// hand-written array carrying Rust function pointers, which meant a new
/// collector's graph needed an edit in a second repo — and a roster that could
/// name a plan for a collector no longer in the registry.
///
/// [`GraphMapping`](crate::graph_mapping::GraphMapping) carries no knut type
/// and no korp type: it is DATA the host compiles.
fn graph_mapping(&self) -> Option<crate::graph_mapping::GraphMapping> {
None
}
/// **Every table mapping this collector declares** — the plural seam, and the
/// one a multi-table source implements.
///
/// [`graph_mapping`](Self::graph_mapping) carries ONE table, which is the right
/// shape for the sources that have one: a landed table whose rows are the
/// entities and whose columns are the edges. `osm-cut` is not that shape — it
/// lands `graph_nodes` and `graph_edges`, vertices in one file and edges in the
/// other.
///
/// That is not an awkward special case. knut's own spec has always been plural
/// (`MappingSpec { tables: Vec<TableMapping> }`), and a graph whose nodes and
/// edges are separate relations is the pattern rel2graph is built for; the
/// single-table declaration was the narrower thing, not the general one.
///
/// The default delegates, so a collector with one mapping implements nothing
/// new and the five that already do are untouched. Every declaration returned
/// here must carry the SAME `plan` — a plan is one graph, and two plans out of
/// one collector would be two pipelines for one thing.
fn graph_mappings(&self) -> Vec<crate::graph_mapping::GraphMapping> {
self.graph_mapping().into_iter().collect()
}
/// **The human-facing card** — who runs this source, who owns the data, where
/// the terms are, how to get a key, how hard it may be called, and what bites.
///
/// `None` means UNDOCUMENTED, and it is a real answer rather than an excuse:
/// `tests/human_docs.rs` fails and NAMES every registered collector that
/// answers `None`, so an undocumented source is a red with a list, not a blank
/// panel an operator discovers on their own.
///
/// Nothing here may be invented — see [`crate::human_doc`].
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
None
}
/// CDC seam: where did we get to? Default: no checkpointing.
async fn checkpoint(&self) -> Result<Option<Checkpoint>> {
Ok(None)
}
}
/// The built-in collector registry — like `ArchiveTypePlugin`'s, it lets the
/// Ingest tab render a dropdown + the test-matrix assert a row per collector.
/// Each is config-driven (base_url/params/auth supplied at [`run`] time).
pub fn registry() -> Vec<Box<dyn Collector>> {
#[allow(unused_mut)]
let mut collectors: Vec<Box<dyn Collector>> = vec![
Box::new(crate::dcat::DcatCrawler),
Box::new(crate::pxweb::PxWebCollector),
Box::new(crate::pxweb::PxWebTableTree),
Box::new(crate::sdmx::SdmxCollector::eurostat()),
Box::new(crate::sdmx::SdmxCollector::oecd()),
Box::new(crate::sdmx::SdmxCollector::ecb()),
Box::new(crate::sdmx::SdmxCollector::insee()),
Box::new(crate::sdmx::SdmxCollector::ons()),
Box::new(crate::sdmx::SdmxCollector::istat()),
Box::new(crate::sdmx::SdmxCollector::worldbank()),
Box::new(crate::entsoe::EntsoeCollector),
Box::new(crate::entsoe_generation::EntsoeGenerationCollector),
Box::new(crate::entsoe_day_ahead::EntsoeDayAheadCollector),
Box::new(crate::nordpool::NordPoolCollector),
Box::new(crate::energy_charts::EnergyChartsCollector),
Box::new(crate::usgs::UsgsQuakeCollector::default()),
Box::new(crate::rss::RssFeedCollector::default()),
Box::new(crate::openaq::OpenAqCollector),
Box::new(crate::gtfs_rt::GtfsRtCollector),
Box::new(crate::rest::RestCollector::polisen()),
Box::new(crate::rest::RestCollector::kolada()),
Box::new(crate::rest::RestCollector::smhi_metobs()),
Box::new(crate::rest::RestCollector::smhi_weather()),
Box::new(crate::rest::RestCollector::digitraffic_ais()),
Box::new(crate::rest::RestCollector::opensky()),
Box::new(crate::rest::RestCollector::opensky_europe_bbox()),
Box::new(crate::rest::RestCollector::opensky_tracks()),
Box::new(crate::rest::RestCollector::aisstream()),
Box::new(crate::rest::RestCollector::gbfs_citybikes()),
Box::new(crate::rest::RestCollector::osm_overpass()),
Box::new(crate::rest::RestCollector::bolagsverket()),
Box::new(crate::rest::RestCollector::gleif_lei()),
Box::new(crate::rest::RestCollector::domstol_rattspraxis()),
Box::new(crate::rest::RestCollector::riksdagen_dokument()),
Box::new(crate::rest::RestCollector::scb_foretagsregistret()),
#[cfg(feature = "csv")]
Box::new(crate::csv::CsvCollector::fi_insyn()),
Box::new(crate::rest::RestCollector::gbif()),
Box::new(crate::rest::RestCollector::worldbank()),
Box::new(crate::rest::RestCollector::restcountries()),
Box::new(crate::rest::RestCollector::smhi_hydro()),
Box::new(crate::rest::RestCollector::metno_forecast()),
Box::new(crate::rest::RestCollector::mobility_database()),
Box::new(crate::rest::RestCollector::eea_wise()),
Box::new(crate::rest::RestCollector::destatis_genesis()),
Box::new(crate::rest::RestCollector::cbs_odata()),
Box::new(crate::rest::RestCollector::mobility_database_rt()),
Box::new(crate::erddap::ErddapCollector::emodnet_physics()),
Box::new(crate::erddap::ErddapCollector::noaa_coastwatch()),
Box::new(crate::wfs::WfsCollector),
Box::new(crate::trafikverket::TrafikverketCollector::train_positions()),
Box::new(crate::trafikverket::TrafikverketCollector::trains()),
Box::new(crate::trafikverket::TrafikverketCollector::road_network()),
];
#[cfg(feature = "osm")]
{
collectors.push(Box::new(crate::osm::OsmGeoCollector));
collectors.push(Box::new(crate::osm::OsmAdminCollector));
}
#[cfg(feature = "gribnc")]
{
collectors.push(Box::new(crate::gribnc::WeatherCollector::copernicus_cds()));
collectors.push(Box::new(crate::gribnc::WeatherCollector::dwd_icon()));
collectors.push(Box::new(crate::gribnc::WeatherCollector::era5()));
collectors.push(Box::new(crate::gribnc::WeatherCollector::cams()));
}
#[cfg(feature = "csv")]
collectors.push(Box::new(crate::csv::CsvCollector::default()));
#[cfg(feature = "gtfs")]
{
collectors.push(Box::new(crate::gtfs::GtfsCollector::stops()));
collectors.push(Box::new(crate::gtfs::GtfsCollector::routes()));
collectors.push(Box::new(crate::gtfs::GtfsCollector::trips()));
}
#[cfg(feature = "mainframe")]
{
collectors.push(Box::new(crate::mainframe::MainframeCollector::default()));
collectors.push(Box::new(crate::fixedwidth::FixedWidthCollector::default()));
}
collectors
}
/// Look up a registered collector by `meta().name`.
pub fn by_name(name: &str) -> Option<Box<dyn Collector>> {
registry().into_iter().find(|c| c.meta().name == name)
}
/// **Every registered collector's REQUIREMENT DECLARATION**, `(name, requirements)`,
/// in registry order — the roster a host renders its collector list from, and the
/// one place "can this run?" is answered for the whole registry. Honours the active
/// feature set exactly as [`registry`] does.
pub fn registry_requirements() -> Vec<(String, crate::requires::Requirements)> {
registry()
.iter()
.map(|c| (c.meta().name, c.requirements()))
.collect()
}
/// Every registered collector's **human card**, by name — what the Collectors pane
/// renders in the detail panel and what `cargo xtask collectors-doc` writes into the
/// manual. `None` for a collector nobody has documented; `tests/human_docs.rs` makes
/// that a named red rather than a blank panel.
pub fn registry_human_docs() -> Vec<(String, Option<crate::human_doc::HumanDoc>)> {
registry()
.iter()
.map(|c| (c.meta().name, c.human_doc()))
.collect()
}
/// Every registered collector's **freshness declaration**, by name — the roster
/// the pane renders a freshness column from and the sweep a test-matrix walks.
pub fn registry_freshness() -> Vec<(String, crate::freshness::Freshness)> {
registry()
.iter()
.map(|c| (c.meta().name, c.freshness()))
.collect()
}
/// Every registered collector's **business key** declaration, by name.
pub fn registry_business_keys() -> Vec<(String, crate::history::BusinessKey)> {
registry()
.iter()
.map(|c| (c.meta().name, c.business_key()))
.collect()
}
/// Enumerate every registered collector as a `(meta, source→target)` row —
/// the discovery surface the test-matrix iterates and the Ingest UI renders.
/// One row per collector makes each an **enumerable, visible test candidate**:
/// iterate `registry_descriptors()` to sweep `(collector, source, target)`
/// tuples. Honours the active feature set (osm/csv/mainframe add rows).
pub fn registry_descriptors() -> Vec<(CollectorMeta, SourceTarget)> {
registry()
.iter()
.map(|c| (c.meta(), c.source_target()))
.collect()
}
/// **Introspection / emit marker** — record one functional-status row for the
/// test matrix. Wraps `nornir_testmatrix::functional_status` behind the
/// `testmatrix` feature (a compiled-out `#[inline]` no-op otherwise, with no
/// nornir dep). `component` is the reporting unit (e.g.
/// `"korp-collectors/dcat"`), `check` what it verified, `ok` the verdict,
/// `detail` a short human note. Collectors + tests call this so `nornir test`
/// SEES each collector.
#[inline]
pub fn functional_status(component: &str, check: &str, ok: bool, detail: &str) {
#[cfg(feature = "testmatrix")]
nornir_testmatrix::functional_status(component, check, ok, detail);
#[cfg(not(feature = "testmatrix"))]
{
let _ = (component, check, ok, detail);
}
}
/// Emit one **descriptor** functional-status row per registered collector
/// (component `"korp-collectors/<name>"`, check `"descriptor_declared"`,
/// detail = the `source → target` label). Drives the *discovery-complete*
/// law: `nornir test --features testmatrix` gets a visible row for every
/// collector even before its data test runs. No-op without `testmatrix`.
pub fn emit_registry_descriptors() {
for (meta, st) in registry_descriptors() {
functional_status(
&format!("korp-collectors/{}", meta.name),
"descriptor_declared",
true,
&st.label(),
);
}
}
/// Path-agnostic runner (behind `skade-sink`): `meta() → schema() →
/// Warehouse::table_or_create → collect() streaming into the Iceberg sink →
/// CollectStats`. This is the korp-lite (Spark-free) landing path; the output
/// is standard Iceberg.
///
/// Returns the collector's [`CollectStats`]; the warehouse row count is the
/// hard end-to-end contract (assert via `Table::count` after this returns).
/// **The one-shot half of the NOT-STARTABLE gate** — refuse before `collect()`,
/// never during it.
///
/// The owner's ruling, 2026-09-06: a collector whose config is not filled in "REFUSES
/// at load with a named reason and never half-starts". On `439b845` the one-shot
/// path half-started by construction: `run_with_sink` opened a sink, called
/// `collect()`, and the collector discovered its own missing key several statements
/// in — `entsoe::collect` builds an HTTP client, then calls
/// `cfg.auth.resolve_async()`, which errors `missing credential env var
/// ENTSOE_TOKEN`. Honest, and still a run that was allowed to begin.
///
/// # The one exception, and why it is not a hole
///
/// A run with [`CollectorCfg::canned`] set is NOT gated. A canned run reads
/// fixtures and never builds a request, so there is no credential for it to be
/// missing — and gating it would make every offline fixture test require the very
/// keys this lane exists to stop asking for. The gate protects the WIRE; a run with
/// no wire has nothing to protect.
///
/// Presence is asked of [`EnvSecretStore`](crate::requires::EnvSecretStore), the
/// leaf's default backend. A host with a vault injects its own store on the daemon
/// path (`CollectorCtx::secret`); this path has no such seam yet, and answering from
/// the environment is what the collectors themselves do a moment later, so the gate
/// and the collector agree.
#[cfg(feature = "parsers")]
fn refuse_unless_startable(collector: &dyn Collector, cfg: &CollectorCfg) -> Result<()> {
if cfg.canned.is_some() {
return Ok(());
}
let effective = collector.requirements().with_source(cfg.base_url.clone());
let store = crate::requires::EnvSecretStore::new();
let presence = crate::requires::StorePresence(&store);
if let Some(reason) = effective.refusal(&presence, &cfg.params) {
return Err(crate::requires::NotStartable {
collector: collector.meta().name,
reason,
}
.into());
}
Ok(())
}
#[cfg(feature = "skade-sink")]
pub async fn run(
collector: &dyn Collector,
cfg: &CollectorCfg,
warehouse: &str,
) -> Result<CollectStats> {
refuse_unless_startable(collector, cfg)?;
let schema = collector.schema(cfg)?;
let wh = skade::Warehouse::open(warehouse).await?;
let table = wh.table_or_create(&cfg.table, &schema).await?;
let sink = BatchSink::iceberg(table);
let mut stats = collector.collect(cfg, Bounds::default(), &sink).await?;
sink.flush().await?;
// The sink's row tally is authoritative (collect() may under/over-report).
stats.rows = sink.rows();
stats.batches = sink.batches();
stats.table = cfg.table.clone();
if stats.rows == 0 && stats.reason.is_none() {
// Annotate, but with the RECOGNISABLE sentinel — `CollectStats::ragnarok`
// reads it back as "still unexplained", so this note can never launder
// an unexplained empty into an honest one.
stats.reason = Some(RAGNAROK_UNEXPLAINED.into());
}
Ok(stats)
}
/// **[`run`] over the HOST's warehouse seam** (feature `daemon`): the collector
/// streams into a [`BatchSink::warehouse`] and `flush` hands every batch to
/// `sink.append(cfg.table, …)` — in korp that is `Streaming::land` on the
/// configured backend row, so a collector lands where the host's Tables view
/// lists. The leaf opens NO warehouse of its own here. `stats.rows` is the
/// number the SINK reported landed, not the push tally.
///
/// This is the runner korp calls for every registry collector; [`run`] (a skade
/// directory the leaf opens itself) stays for standalone consumers and the
/// fixture tests.
#[cfg(feature = "daemon")]
pub async fn run_with_sink(
collector: &dyn Collector,
cfg: &CollectorCfg,
sink: std::sync::Arc<dyn crate::daemon::WarehouseSink>,
) -> Result<CollectStats> {
refuse_unless_startable(collector, cfg)?;
// `schema()` is still consulted: a collector that cannot describe itself
// must not stream, and the host seam creates the table from the batch's
// own schema on first append.
let _ = collector.schema(cfg)?;
let bsink = BatchSink::warehouse(sink, cfg.table.clone());
let mut stats = collector.collect(cfg, Bounds::default(), &bsink).await?;
bsink.flush().await?;
stats.rows = bsink.landed();
stats.batches = bsink.batches();
stats.table = cfg.table.clone();
if stats.rows == 0 && stats.reason.is_none() {
stats.reason = Some(RAGNAROK_UNEXPLAINED.into());
}
Ok(stats)
}
/// **What one incremental run did** — the stats the sink reported plus the
/// freshness verdict, so a caller can tell "nothing changed" from "nothing landed".
#[derive(Debug, Clone)]
pub struct IncrementalRun {
/// The ordinary collect stats.
pub stats: CollectStats,
/// The plan this run was given.
pub plan: crate::freshness::Plan,
/// What it turned out to be.
pub delta: crate::freshness::Delta,
/// The mark that was PERSISTED, or `None` when nothing was written.
pub advanced: Option<crate::freshness::ReadMark>,
}
impl IncrementalRun {
/// The one-line report: plan → outcome.
pub fn line(&self) -> String {
match &self.delta {
crate::freshness::Delta::Unchanged { how, bytes } => {
format!("{} → unchanged ({how}), {bytes} bytes", self.plan.line())
}
crate::freshness::Delta::Landed { rows, bytes, mark, .. } => format!(
"{} → {rows} rows, {bytes} bytes, mark now {}",
self.plan.line(),
if mark.is_empty() { "(validators)" } else { mark }
),
}
}
}
/// **The default incremental behaviour, in one place.**
///
/// 1. ask the collector for its [`Freshness`](crate::freshness::Freshness),
/// 2. resolve the stored [`ReadMark`](crate::freshness::ReadMark) for
/// `(collector, source)`,
/// 3. derive the [`Plan`](crate::freshness::Plan) and hand it down as [`Bounds`],
/// 4. land,
/// 5. **advance the mark only after the land succeeded.**
///
/// Step 5 is the whole safety property and it is why this function exists rather
/// than a note in a design doc: an error anywhere before `advance` returns `Err`
/// with the previous mark untouched, so the next run REDOES the slice. Re-landing
/// rows is a duplicate the history columns absorb; a skipped slice is a hole nobody
/// ever notices.
///
/// Every collector inherits this — the freshness declaration is what varies, not
/// the runner.
#[cfg(feature = "daemon")]
pub async fn run_incremental(
collector: &dyn Collector,
cfg: &CollectorCfg,
sink: std::sync::Arc<dyn crate::daemon::WarehouseSink>,
marks: &dyn crate::freshness::MarkStore,
) -> Result<IncrementalRun> {
// The gate first: an unconfigured collector must not even move a freshness
// mark. A refused run that had already read a mark would look, to the next run,
// exactly like a run that found nothing new.
refuse_unless_startable(collector, cfg)?;
let name = collector.meta().name;
let fresh = collector.freshness();
let held = marks.mark(&name, &cfg.base_url);
let plan = crate::freshness::Plan::of(&fresh, held.as_ref());
let _ = collector.schema(cfg)?;
let bsink = BatchSink::warehouse(sink, cfg.table.clone());
let mut stats = collector
.collect(cfg, Bounds::plan(plan.clone()), &bsink)
.await?;
// The FLUSH is what makes the rows durable, so it must succeed before the mark
// is allowed to move. `?` here leaves the previous mark in place.
bsink.flush().await?;
stats.rows = bsink.landed();
stats.batches = bsink.batches();
stats.table = cfg.table.clone();
let ck = collector.checkpoint().await?.unwrap_or_default();
let how = stats
.reason
.as_deref()
.filter(|r| crate::freshness::is_unchanged_reason(Some(r)))
.unwrap_or("no new records")
.to_string();
let delta = ck.delta(stats.rows, &how);
// An unexplained empty is still RED — an incremental run does not get to
// launder one, and a collector that forgot its `reason` must not be given the
// "unchanged" excuse.
//
// **And a run that held nothing back cannot HAVE an unchanged excuse.** A
// `QueryWindow` collector on its first tick plans `Window { since: None }`,
// asks for everything, and — if the source answers with an empty array —
// reports "…returned no records". That is a description, not an explanation:
// nothing was excluded, so the source is empty, which is exactly what
// empty-is-RED exists to catch. `Plan::resumed` is the one writer of that
// distinction, so the reason a collector offers is accepted only when the plan
// actually carried a mark, a validator, an offset or a version.
if stats.rows == 0 && (stats.reason.is_none() || !plan.resumed()) {
stats.reason = Some(RAGNAROK_UNEXPLAINED.into());
}
// ONLY NOW. Every path above returns before this line on failure.
let advanced = match delta.next_mark(&fresh, held.as_ref()) {
Some(m) if stats.is_red() => {
// A red run advances nothing: a collector that landed nothing and
// could not say why has not proven it read anything.
let _ = m;
None
}
Some(m) => {
marks.advance(&name, &cfg.base_url, &m)?;
Some(m)
}
None => None,
};
Ok(IncrementalRun {
stats,
plan,
delta,
advanced,
})
}
/// [`run`] with the RAGNARÖK gate ENFORCED: an unexplained 0-row collect
/// (see [`CollectStats::ragnarok`]) comes back as `Err` instead of a
/// deceptively-`Ok` empty. `run()` stays as it was for callers that want to
/// inspect the stats themselves; this is the one-call form for a scheduler or
/// CI job that must go RED rather than record a silent no-op.
///
/// A 0-row collect the COLLECTOR explained (`reason` set by `collect()`) is
/// still `Ok` — RAGNARÖK forbids the silent empty, not the honest one.
#[cfg(feature = "skade-sink")]
pub async fn run_strict(
collector: &dyn Collector,
cfg: &CollectorCfg,
warehouse: &str,
) -> Result<CollectStats> {
let stats = run(collector, cfg, warehouse).await?;
stats.ragnarok_check()?;
Ok(stats)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::Direction;
use crate::sink::BatchSink;
use anyhow::Result as AnyResult;
use arrow_schema::Schema as ArrowSchema;
/// **The registry-wide gate.** Every registered collector answers the
/// requirement question, and the answers DISCRIMINATE — the roster contains
/// collectors in more than one state, so a `requirements()` that returned the
/// same thing for everyone (or an assessor stuck on one arm) goes red here.
#[test]
fn every_registered_collector_declares_what_it_needs_and_the_states_differ() {
use crate::requires::{NoSecrets, Readiness};
let reqs = registry_requirements();
assert!(reqs.len() >= 40, "the roster is the whole registry: {}", reqs.len());
let states: Vec<(String, Readiness)> = reqs
.iter()
.map(|(n, r)| {
(n.clone(), r.readiness(&NoSecrets, &serde_json::json!({}), true, false))
})
.collect();
let runnable = states.iter().filter(|(_, s)| s.key() == "runnable").count();
let blocked = states.iter().filter(|(_, s)| s.key() == "needs_config").count();
assert!(runnable > 0, "some collectors need nothing and ARE runnable");
assert!(blocked > 0, "some collectors are honestly blocked");
assert_eq!(runnable + blocked, states.len(), "no third answer without a live fact");
// Every blocked collector NAMES what it is missing — an unnamed refusal is
// a dead end for the operator, and is the state this work replaced.
for (n, s) in &states {
if let Readiness::NeedsConfig { missing: m } = s {
assert!(!m.is_empty(), "`{n}` is blocked for no stated reason");
for x in m {
assert!(!x.line().trim().is_empty(), "`{n}` has a blank reason");
}
}
}
}
/// **The two collectors the owner named, by name.** `opensky-flights` is blocked
/// on the OpenSky OAuth2 pair BEFORE a request is built (it used to reach the
/// wire and answer `korp-rpc-21: GET → builder error`); `polisen-events` needs
/// no key and is runnable. Both arms, so neither can be a constant.
#[test]
fn opensky_is_blocked_by_name_and_polisen_is_runnable() {
use crate::requires::{HaveSecrets, NoSecrets};
let reqs = registry_requirements();
let get = |n: &str| {
reqs.iter().find(|(k, _)| k == n).map(|(_, r)| r.clone()).unwrap_or_else(|| panic!("`{n}` is registered"))
};
let sky = get("opensky-flights");
let blocked = sky.readiness(&NoSecrets, &serde_json::json!({}), true, false);
assert_eq!(blocked.key(), "needs_config");
let refusal = blocked.refusal().unwrap();
assert!(refusal.contains("OPENSKY_CLIENT_ID"), "{refusal}");
assert!(refusal.contains("OPENSKY_CLIENT_SECRET"), "{refusal}");
// Its SOURCE was never the problem — the endpoint is declared and real.
assert!(sky.source.starts_with("https://opensky-network.org/"), "{}", sky.source);
// With both names supplied it becomes runnable — the guard is not a constant.
assert_eq!(
sky.readiness(
&HaveSecrets::of(&["OPENSKY_CLIENT_ID", "OPENSKY_CLIENT_SECRET"]),
&serde_json::json!({}),
true,
false
)
.key(),
"runnable"
);
// …and half the pair is still a refusal that names the half that is missing.
let half = sky.readiness(
&HaveSecrets::of(&["OPENSKY_CLIENT_ID"]),
&serde_json::json!({}),
true,
false,
);
assert_eq!(half.key(), "needs_config");
let hr = half.refusal().unwrap();
assert!(hr.contains("OPENSKY_CLIENT_SECRET") && !hr.contains("OPENSKY_CLIENT_ID is not set"), "{hr}");
let pol = get("polisen-events");
assert_eq!(pol.readiness(&NoSecrets, &serde_json::json!({}), true, false).key(), "runnable");
assert!(pol.secrets.is_empty(), "polisen needs no key");
// trafikverket-trainpos: honestly blocked on ONE named key.
let tv = get("trafikverket-trainpos");
let tvr = tv.readiness(&NoSecrets, &serde_json::json!({}), true, false);
assert_eq!(tvr.key(), "needs_config");
assert!(tvr.refusal().unwrap().contains("TRAFIKVERKET_API_KEY"));
assert_eq!(
tv.readiness(&HaveSecrets::of(&["TRAFIKVERKET_API_KEY"]), &serde_json::json!({}), true, false).key(),
"runnable"
);
}
/// An out-of-tree `Collector` that DOES NOT override `source_target()` —
/// it exercises the trait's DEFAULT derivation (the "still compiles"
/// contract). Parameterised by `source_kind`.
struct MockCollector {
name: &'static str,
kind: SourceKind,
}
#[async_trait::async_trait]
impl Collector for MockCollector {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: self.name.into(),
direction: Direction::Source,
source_kind: self.kind,
path: CollectorPath::RustNative,
formats: vec![],
license: "test".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec![],
description: "mock".into(),
}
}
fn schema(&self, _cfg: &CollectorCfg) -> AnyResult<ArrowSchema> {
Ok(ArrowSchema::empty())
}
async fn collect(
&self,
_cfg: &CollectorCfg,
_bounds: Bounds,
_sink: &BatchSink,
) -> AnyResult<CollectStats> {
Ok(CollectStats::default())
}
}
fn mock(kind: SourceKind) -> SourceTarget {
MockCollector { name: "mock", kind }.source_target()
}
#[test]
fn default_source_target_derives_endpoint_from_source_kind() {
assert_eq!(mock(SourceKind::Dcat).source, SourceEndpoint::DcatCatalog);
assert_eq!(mock(SourceKind::PxWeb).source, SourceEndpoint::PxWebApi);
// Rest / Ogc / Gtfs all collapse to the generic REST endpoint.
assert_eq!(mock(SourceKind::Rest).source, SourceEndpoint::GenericRest);
assert_eq!(mock(SourceKind::Ogc).source, SourceEndpoint::GenericRest);
assert_eq!(mock(SourceKind::Gtfs).source, SourceEndpoint::GenericRest);
// Every file-shaped family falls to FlatFile.
assert_eq!(mock(SourceKind::BulkFile).source, SourceEndpoint::FlatFile);
assert_eq!(mock(SourceKind::FlatFile).source, SourceEndpoint::FlatFile);
assert_eq!(
mock(SourceKind::Spreadsheet).source,
SourceEndpoint::FlatFile
);
assert_eq!(mock(SourceKind::Geo).source, SourceEndpoint::FlatFile);
}
#[test]
fn default_source_target_is_a_warehouse_with_no_endpoint_and_name_table() {
let st = mock(SourceKind::Rest);
assert_eq!(st.target, TargetKind::WarehouseTable);
assert!(st.map_scene.is_none());
assert_eq!(
st.source_endpoint, "",
"default carries no representative endpoint"
);
assert_eq!(
st.target_table, "mock",
"default lands in a table named after the collector"
);
}
#[test]
fn source_target_constructors_set_fields() {
let wh = SourceTarget::warehouse(SourceEndpoint::PxWebApi, "https://api", "cube");
assert_eq!(wh.target, TargetKind::WarehouseTable);
assert_eq!(wh.source_endpoint, "https://api");
assert_eq!(wh.target_table, "cube");
assert!(wh.map_scene.is_none());
let ms =
SourceTarget::map_scene(SourceEndpoint::Ais, "https://feed", "pos", "osm3d-flight");
assert_eq!(ms.target, TargetKind::MapScene);
assert_eq!(ms.map_scene.as_deref(), Some("osm3d-flight"));
}
#[test]
fn label_format_for_warehouse_and_map_scene() {
let wh =
SourceTarget::warehouse(SourceEndpoint::DcatCatalog, "https://x", "dcat_registry");
assert_eq!(wh.label(), "dcat-catalog → warehouse-table (dcat_registry)");
let ms = SourceTarget::map_scene(
SourceEndpoint::Ais,
"https://x",
"ais_positions",
"osm3d-flight",
);
assert_eq!(
ms.label(),
"ais → map-scene (ais_positions, scene=osm3d-flight)"
);
}
#[test]
fn source_and_target_tags_are_stable_and_distinct() {
use std::collections::HashSet;
let sources = [
SourceEndpoint::DcatCatalog,
SourceEndpoint::PxWebApi,
SourceEndpoint::Eurostat,
SourceEndpoint::EntsoE,
SourceEndpoint::Trafikverket,
SourceEndpoint::Ais,
SourceEndpoint::Osm,
SourceEndpoint::OsmAdmin,
SourceEndpoint::GenericRest,
SourceEndpoint::FlatFile,
SourceEndpoint::Wfs,
SourceEndpoint::Gtfs,
SourceEndpoint::GtfsRt,
SourceEndpoint::Bolagsverket,
SourceEndpoint::NordPool,
SourceEndpoint::Usgs,
SourceEndpoint::Feed,
SourceEndpoint::OpenAq,
SourceEndpoint::EnergyCharts,
];
let tags: HashSet<&str> = sources.iter().map(|s| s.as_str()).collect();
assert_eq!(
tags.len(),
sources.len(),
"every SourceEndpoint tag is distinct"
);
let targets = [
TargetKind::WarehouseTable,
TargetKind::MapScene,
TargetKind::Arrow,
];
let ttags: HashSet<&str> = targets.iter().map(|t| t.as_str()).collect();
assert_eq!(
ttags.len(),
targets.len(),
"every TargetKind tag is distinct"
);
}
#[test]
fn source_target_serde_round_trips() {
let st = SourceTarget::map_scene(
SourceEndpoint::Trafikverket,
"https://api.trafikinfo.trafikverket.se/v2/data.json",
"trafikverket-trainpos",
"osm3d-flight",
);
let json = serde_json::to_string(&st).expect("serialize");
let back: SourceTarget = serde_json::from_str(&json).expect("deserialize");
assert_eq!(back.source, st.source);
assert_eq!(back.target, st.target);
assert_eq!(back.source_endpoint, st.source_endpoint);
assert_eq!(back.target_table, st.target_table);
assert_eq!(back.map_scene, st.map_scene);
}
#[test]
fn auth_config_resolves_present_and_errors_on_missing_env() {
// Resolve → live Auth → apply to a request → inspect what it injected.
fn built(auth: &dyn crate::http::Auth) -> reqwest::Request {
let req = crate::tls_roots(reqwest::Client::builder()).build().expect("reqwest client").get("http://example.test/");
auth.apply(req).build().expect("build request")
}
// None is always Ok and injects nothing.
let none = AuthConfig::None.resolve().expect("None resolves");
let r = built(none.as_ref());
assert_eq!(r.url().query(), None);
assert!(r.headers().get(reqwest::header::AUTHORIZATION).is_none());
// QueryKey resolves from env and appends `?param=value`.
unsafe { std::env::set_var("KORP_TEST_QK", "QSECRET") };
let qk = AuthConfig::QueryKey {
param: "securityToken".into(),
env: "KORP_TEST_QK".into(),
}
.resolve()
.expect("QueryKey resolves when env present");
assert_eq!(
built(qk.as_ref()).url().query(),
Some("securityToken=QSECRET")
);
// HeaderKey resolves and sets the header.
unsafe { std::env::set_var("KORP_TEST_HK", "HSECRET") };
let hk = AuthConfig::HeaderKey {
header: "X-Api-Key".into(),
env: "KORP_TEST_HK".into(),
}
.resolve()
.expect("HeaderKey resolves");
assert_eq!(
built(hk.as_ref()).headers().get("X-Api-Key").unwrap(),
"HSECRET"
);
// Bearer resolves to `Authorization: Bearer <token>`.
unsafe { std::env::set_var("KORP_TEST_BR", "BSECRET") };
let br = AuthConfig::Bearer {
env: "KORP_TEST_BR".into(),
}
.resolve()
.expect("Bearer resolves");
assert_eq!(
built(br.as_ref())
.headers()
.get(reqwest::header::AUTHORIZATION)
.unwrap(),
"Bearer BSECRET"
);
// A credentialled config whose env var is ABSENT errors, naming the var.
unsafe { std::env::remove_var("KORP_TEST_ABSENT") };
// (Ok payload is `Box<dyn Auth>`, not Debug, so match rather than expect_err.)
let err = match (AuthConfig::Bearer {
env: "KORP_TEST_ABSENT".into(),
})
.resolve()
{
Ok(_) => panic!("absent credential env must error"),
Err(e) => e,
};
assert!(
err.to_string().contains("KORP_TEST_ABSENT"),
"error names the missing var: {err}"
);
unsafe {
std::env::remove_var("KORP_TEST_QK");
std::env::remove_var("KORP_TEST_HK");
std::env::remove_var("KORP_TEST_BR");
}
}
#[test]
fn oauth2_client_credentials_sync_resolve_directs_to_async() {
// OAuth2 needs a network token fetch → the sync resolver refuses and
// names the async path (a mis-wired sync call fails loudly, not silently).
let cfg = AuthConfig::OAuth2ClientCredentials {
token_url: "https://auth.opensky-network.org/token".into(),
id_env: "OPENSKY_CLIENT_ID".into(),
secret_env: "OPENSKY_CLIENT_SECRET".into(),
};
let err = match cfg.resolve() {
Ok(_) => panic!("OAuth2 must not resolve synchronously"),
Err(e) => e,
};
assert!(
err.to_string().contains("resolve_async"),
"sync resolve points to the async path: {err}"
);
}
#[tokio::test]
async fn oauth2_resolve_async_errors_on_missing_creds_naming_the_var() {
unsafe {
std::env::remove_var("KORP_OAUTH_ID_ABSENT");
std::env::remove_var("KORP_OAUTH_SECRET_ABSENT");
}
let cfg = AuthConfig::OAuth2ClientCredentials {
token_url: "https://auth.example/token".into(),
id_env: "KORP_OAUTH_ID_ABSENT".into(),
secret_env: "KORP_OAUTH_SECRET_ABSENT".into(),
};
let err = match cfg.resolve_async().await {
Ok(_) => panic!("absent OAuth2 client id must error before any network call"),
Err(e) => e,
};
assert!(
err.to_string().contains("KORP_OAUTH_ID_ABSENT"),
"error names the missing credential var: {err}"
);
}
#[tokio::test]
async fn resolve_async_delegates_non_oauth_variants_to_sync() {
// A non-OAuth config (None) resolves identically through resolve_async.
let none = AuthConfig::None
.resolve_async()
.await
.expect("None resolves async");
let req = crate::tls_roots(reqwest::Client::builder()).build().expect("reqwest client").get("http://example.test/");
let built = none.apply(req).build().expect("build request");
assert_eq!(
built.url().query(),
None,
"None injects nothing via resolve_async"
);
}
// ── RAGNARÖK verdict: empty is RED, never silently green ─────────────
/// The three verdicts, and the one that matters: a 0-row collect with no
/// reason is RED. RED-when-broken — if `ragnarok()` ever starts calling an
/// unexplained empty honest, this fails.
#[test]
fn ragnarok_verdict_separates_rows_explained_empty_and_red() {
let rows = CollectStats {
rows: 7,
batches: 1,
table: "t".into(),
reason: None,
};
assert_eq!(rows.ragnarok(), Ragnarok::Rows);
assert!(!rows.is_red());
let explained = CollectStats {
rows: 0,
reason: Some("window provably empty (no trading day)".into()),
..Default::default()
};
assert_eq!(explained.ragnarok(), Ragnarok::ExplainedEmpty);
assert!(!explained.is_red(), "an EXPLAINED empty is honest, not RED");
let silent = CollectStats {
rows: 0,
table: "quakes".into(),
..Default::default()
};
assert_eq!(silent.ragnarok(), Ragnarok::Red);
assert!(silent.is_red(), "0 rows + no reason is RED (RAGNARÖK)");
let err = match silent.ragnarok_check() {
Ok(_) => panic!("an unexplained empty must not check out green"),
Err(e) => e.to_string(),
};
assert!(err.contains("quakes"), "the error names the table: {err}");
assert!(err.contains("RAGNARÖK"), "the error names the law: {err}");
}
/// **The laundering hole.** `run()` stamps [`RAGNAROK_UNEXPLAINED`] onto a
/// 0-row collect the collector never explained. That annotation must NOT
/// count as an explanation — otherwise the runner turns every silent empty
/// green. RED-when-broken: if the sentinel is ever treated as a real
/// reason, this flips to `ExplainedEmpty` and fails.
#[test]
fn runner_placeholder_reason_does_not_launder_an_unexplained_empty() {
let laundered = CollectStats {
rows: 0,
table: "t".into(),
reason: Some(RAGNAROK_UNEXPLAINED.into()),
..Default::default()
};
assert_eq!(
laundered.ragnarok(),
Ragnarok::Red,
"the runner's own placeholder is not an explanation"
);
assert!(laundered.is_red());
assert!(laundered.ragnarok_check().is_err());
// ...while ANY collector-authored reason is.
let honest = CollectStats {
reason: Some("source served an empty document".into()),
..Default::default()
};
assert_eq!(honest.ragnarok(), Ragnarok::ExplainedEmpty);
}
#[test]
fn ragnarok_tags_are_stable_and_distinct() {
use std::collections::HashSet;
let all = [Ragnarok::Rows, Ragnarok::ExplainedEmpty, Ragnarok::Red];
let tags: HashSet<&str> = all.iter().map(|r| r.as_str()).collect();
assert_eq!(tags.len(), all.len());
assert_eq!(Ragnarok::Red.as_str(), "red");
}
#[test]
fn registry_descriptors_invariants_hold_for_every_builtin() {
let descriptors = registry_descriptors();
assert!(descriptors.len() >= 8, "≥8 default collectors");
for (meta, st) in &descriptors {
// Network-backed built-ins name a representative endpoint; the
// flat-file collectors (local files, no URL) legitimately don't.
assert!(
!st.source_endpoint.is_empty() || st.source == SourceEndpoint::FlatFile,
"{} names a representative endpoint unless it is a flat file",
meta.name
);
assert!(
!st.target_table.is_empty(),
"{} names a target table",
meta.name
);
// The map_scene field is present iff the target is a map scene.
assert_eq!(
st.map_scene.is_some(),
st.target == TargetKind::MapScene,
"{} map_scene must be Some iff target is MapScene",
meta.name
);
// label() mentions both halves.
assert!(st.label().contains(st.source.as_str()));
assert!(st.label().contains(st.target.as_str()));
}
}
}
}
#[cfg(feature = "parsers")]
pub use surface::*;
// ── connector implementations (parsers = light pure-Rust; osm = heavy geo) ──
#[cfg(feature = "parsers")]
pub mod arrow_util;
#[cfg(feature = "parsers")]
pub mod dcat;
/// Energy-Charts (Fraunhofer ISE) public-power generation mix — the parallel
/// `unix_seconds[]` time axis ZIPPED against each `production_types[].data[]` by
/// index into per-`(time, type)` power rows → Arrow. A dedicated transpose of a
/// columnar document (distinct from every other collector: no other one recovers
/// its value by index-aligning two independent parallel arrays).
#[cfg(feature = "parsers")]
pub mod energy_charts;
#[cfg(feature = "parsers")]
pub mod entsoe;
/// ENTSO-E day-ahead electricity SPOT PRICES (`documentType=A44`) — the energy
/// MARKET (price) sibling of [`entsoe`]/[`entsoe_generation`] (volumes): reads
/// `price.amount` per zone × hour → Arrow. The open (free-token) route to
/// day-ahead prices; the licensed twin is [`nordpool`].
#[cfg(feature = "parsers")]
pub mod entsoe_day_ahead;
#[cfg(feature = "parsers")]
pub mod entsoe_generation;
/// GTFS-Realtime `VehiclePositions` — a protobuf `FeedMessage` decoded by a
/// hand-rolled varint/length-delimited reader (no `prost`/`protobuf` dep) into a
/// live transit vehicle-position table (entity/vehicle/route/trip id + lat/lon +
/// bearing/speed/timestamp/stop). A binary wire shape NO other collector decodes
/// (all others are XML/JSON/CSV); the static GTFS zip is [`gtfs`], this is its
/// real-time protobuf companion.
#[cfg(feature = "parsers")]
pub mod gtfs_rt;
#[cfg(feature = "parsers")]
pub mod http;
/// Nord Pool day-ahead electricity spot prices — the `multiAreaEntries`
/// period×zone matrix flattened to per-`(area, period)` price rows (a dedicated
/// flattener; the variable per-area map keys don't fit the generic REST adapter).
#[cfg(feature = "parsers")]
pub mod nordpool;
/// OpenAQ open air-quality measurements — each `results[]` reading flattened to a
/// measurement row (station, parameter, value, unit, lon/lat, UTC time) via a
/// pure-Rust `serde_json` nested-object (coordinates + date) flatten → Arrow. A
/// dedicated flattener, distinct from USGS's GeoJSON geometry decode, the
/// statistical cubes (PxWeb/Eurostat), and the generic REST field-map.
#[cfg(feature = "parsers")]
pub mod openaq;
#[cfg(feature = "parsers")]
pub mod pxweb;
#[cfg(feature = "parsers")]
pub mod rest;
/// Offline-safe **real-or-fake** sample data (`*_try_get_real_sample_data`): each
/// collector's sample source ATTEMPTS a short-timeout live fetch and FALLS BACK
/// to an embedded synthetic fixture flagged `Synthetic` — so tests + benches
/// never hang and never fail purely for want of a network.
/// RSS 2.0 / Atom 1.0 syndication feeds — each `item`/`entry` flattened to a row
/// (feed_title, id, title, link, summary, author, published, categories JSON) via
/// a pure-Rust `quick-xml` parse. A dedicated dialect-agnostic flattener, distinct
/// from the WFS GML feature parser and every JSON collector.
#[cfg(feature = "parsers")]
pub mod rss;
#[cfg(feature = "parsers")]
pub mod sample;
/// Eurostat SDMX-JSON 2.1 statistics — the series/observation dimension split
/// flattened to per-`(series dims × observation dims)` value rows → Arrow. A
/// dedicated flattener (distinct from PxWeb's JSON-stat2 cube).
#[cfg(feature = "parsers")]
pub mod sdmx;
#[cfg(feature = "parsers")]
pub mod sink;
#[cfg(feature = "parsers")]
pub mod trafikverket;
/// USGS real-time earthquake **GeoJSON** `FeatureCollection` → per-Point quake rows
/// (id, place, magnitude, depth, lon/lat, epoch-ms time) → Arrow. A pure-Rust
/// `serde_json` geometry/property flatten — distinct from PxWeb's JSON-stat2 cube,
/// Eurostat's SDMX-JSON split, and the WFS GML parser.
#[cfg(feature = "parsers")]
pub mod usgs;
/// OGC WFS `GetFeature` GML `FeatureCollection` → feature rows (WKT geometry +
/// properties JSON) via a pure-Rust `quick-xml` GML parse (no geo engine).
#[cfg(feature = "parsers")]
pub mod wfs;
/// ERDDAP tabledap / griddap scientific-data server (EMODnet + global
/// ocean/atmosphere ERDDAP deployments) — the self-describing `.json` `table`
/// document (`columnNames` + `columnTypes` + `rows` matrix) decoded to typed
/// Arrow via a pure-Rust `serde_json` parse (no new deps). Uniform over ANY
/// dataset id on ANY ERDDAP server; presets a couple of known EMODnet + NOAA
/// servers. The `.csv` variant is reachable through the generic [`csv`] collector.
#[cfg(feature = "parsers")]
pub mod erddap;
/// CSV (RFC 4180 + real-world dialects: `;`/TAB, BOM, Latin-1, decimal-comma) →
/// typed Arrow via a two-pass schema sniff. Behind `--features csv`.
#[cfg(feature = "csv")]
pub mod csv;
/// Fixed-width flat files + the shared COBOL-copybook layout parser (PIC/USAGE →
/// Arrow schema, record slicer, ASCII decode). Behind `--features mainframe`.
#[cfg(feature = "mainframe")]
pub mod fixedwidth;
/// Mainframe EBCDIC + COBOL-copybook + COMP-3 decode → Arrow (schema from the
/// copybook). Behind `--features mainframe`.
#[cfg(feature = "mainframe")]
pub mod mainframe;
/// GTFS static transit feeds — a pure-Rust ZIP reader (central-directory walk,
/// STORED + DEFLATE) + the `csv` reader per known member → typed stops/routes/trips
/// Arrow tables. Behind `--features gtfs` (pulls `flate2` + `csv`).
#[cfg(feature = "gtfs")]
pub mod gtfs;
/// Administrative-boundaries facet — OSM `boundary=administrative` → an
/// `admin_units` border table + a world admin-division trie. Pure tag-filter +
/// trie (testable without a PBF); the PBF read is in [`osm`].
#[cfg(feature = "parsers")]
pub mod osm_admin;
/// OSM PBF/XML → WKB geometry via znippy-zoomies/osm-katana → skade. Behind
/// `--features osm` (pulls the heavier osm-katana geo crate).
#[cfg(feature = "osm")]
pub mod osm;
/// NetCDF/GRIB weather-model collector — Copernicus CDS + DWD ICON open data.
/// Hand-rolled PURE-RUST GRIB2 + NetCDF-classic decoders (zero new deps) → a
/// variable/time/bbox metadata catalog AND the data-array unpack → Arrow. GRIB2
/// Section 5/6/7 decodes simple (5.0) + complex (5.2) + complex-with-spatial-
/// differencing (5.3) packing with Section-6 bitmaps; NetCDF-classic reads
/// variable data arrays (fixed + record dims). NetCDF-4/HDF5 stays deferred.
/// Behind `--features gribnc`.
#[cfg(feature = "gribnc")]
pub mod gribnc;
/// The **znippy converter** — pack collected Arrow batches / raw byte blobs into a
/// znippy archive (compact, checksummed storage/transport) and read them back
/// byte-identical, reusing the `znippy-common` pack format. Behind `--features
/// znippy`.
#[cfg(feature = "znippy")]
pub mod znippy_pack;
// The `region` module (`RegionSource` / `HttpOsmSource` / `CachedLocalSource`) was
// DELETED 2026-08-13. It fetched `{base_url}/{region}-latest.osm.pbf` Geofabrik-style
// and had ZERO callers in korp or korp-collectors — only these doc lines. korp does not
// acquire regions over HTTP at all: `korp::download` pulls the OSM planet over
// BitTorrent (osm-katana/librqbit) and `korp::pipeline` cuts every region out of it
// locally (convert pass 1 → pass 2 → relation clip → tiling). Geofabrik is not an OSM
// source in this house.
#[cfg(feature = "parsers")]
pub use sink::BatchSink;
/// reqwest 0.13 with reqwest 0.12 `rustls-tls` behaviour (2026-10-02): the ring
/// provider and the bundled Mozilla roots (webpki), not 0.13's aws-lc-rs + OS
/// trust store. Roots added later on the builder are merged in, as in 0.12.
#[cfg(feature = "parsers")]
#[allow(dead_code)]
fn tls_roots(b: reqwest::ClientBuilder) -> reqwest::ClientBuilder {
let _ = rustls::crypto::ring::default_provider().install_default();
b.tls_certs_only(
webpki_root_certs::TLS_SERVER_ROOT_CERTS
.iter()
.filter_map(|c| reqwest::Certificate::from_der(c).ok()),
)
}