src/iceberg_daemons.rs
//! **WAVE 1 — the Iceberg-sink collector daemons** (feature `daemon`, design §11).
//!
//! [`osm_cut`](crate::osm_cut) proved the [`daemon`](crate::daemon) architecture
//! for an **S3-sink** collector (planet → tiles → a `nordisk-maps` bucket). This
//! module adopts the first three EXISTING collectors — **`osm-admin`**, **`dcat`**,
//! **`pxweb`** — to the same architecture as **warehouse-sink** daemons: they FETCH an
//! external open-data source, parse it with the crate's existing pure-Rust parsers
//! ([`crate::osm_admin`] / [`crate::dcat`] / [`crate::pxweb`]), and land the parsed
//! Arrow [`RecordBatch`] through the [`WarehouseSink`](crate::daemon::WarehouseSink)
//! seam into their Iceberg table. The only thing that differs from `osm-cut` is
//! which [`CollectorCtx`] seam they use (the `warehouse` + `fetch` seams instead of
//! the `bucket` seam) — the state machine, registry, Settings + matrix wiring are all
//! generic. The remaining ~17 collectors are the mechanical follow-on.
//!
//! ## The shared state machine (design §3, reused)
//! ```text
//! Offline → Downloading(fetch %) → Indexing(parse) → Online // rows landed
//! → Failed{reason} // empty-is-RED
//! ```
//! Being pre-Online is a normal fact; an **enabled collector that fetches+parses to
//! ZERO rows is a loud [`Phase::Failed`]** (design "empty-is-RED": an enabled collector
//! producing nothing is a failure, never a silent green).
//!
//! ## Offline-safe headless proof
//! The [`fetch`](CollectorCtx::fetch) seam means the run-loop + sink + matrix wiring is
//! proven with **no live gov API** — a test wires a canned [`Fetcher::from_map`] fixture
//! (the crate's `sample::SYNTHETIC_*` bytes) so `reconcile_once` drives all the way to
//! ONLINE deterministically. korp wires the real rustls HTTP core behind the same seam.
use std::sync::{Arc, Mutex};
use anyhow::{Context, Result, anyhow};
use arrow_array::RecordBatch;
use crate::daemon::{
Collector, CollectorConfig, CollectorCtx, CollectorState, ConfigField, ConfigSchema, Phase,
};
// ─────────────────────────────────────────────────────────────────────────────
// Shared run state + the generic warehouse-sink reconcile skeleton.
// ─────────────────────────────────────────────────────────────────────────────
/// The mutable inner run state shared by every warehouse-sink daemon — the daemon
/// thread updates it; the infra matrix reads it (via [`CollectorState::facet`]).
#[derive(Clone, Debug, Default)]
pub struct IcebergRun {
/// The current state-machine value.
pub phase: Phase,
/// The Iceberg table rows land in.
pub table: String,
/// Rows landed in the last successful reconcile.
pub rows_written: u64,
/// Unix seconds of the last successful ONLINE.
pub last_run_unix: u64,
/// The last failure reason, if any.
pub error: Option<String>,
}
fn now_unix() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
/// **How a warehouse-sink daemon acquires its source bytes** — a plain GET, or a
/// query POSTed in the request body (Trafikverket's XML `<REQUEST>` carrying the
/// authentication key). Both ride the one [`Fetcher`](crate::daemon::Fetcher) seam and
/// the one [`reconcile_iceberg`] body, so a POST-body source is not a second daemon
/// skeleton — only its request shape differs. The offline [`Fetcher::from_map`] fixture
/// serves either verb by URL, so both are proven headless.
#[derive(Clone, Debug)]
pub enum Source {
/// GET `url`.
Get { url: String },
/// POST `body` (with `content_type`) to `url`.
Post {
url: String,
body: String,
content_type: String,
},
}
impl Source {
/// A GET source.
pub fn get(url: impl Into<String>) -> Self {
Source::Get { url: url.into() }
}
/// A POST source (query-in-the-body, e.g. Trafikverket XML).
pub fn post(
url: impl Into<String>,
body: impl Into<String>,
content_type: impl Into<String>,
) -> Self {
Source::Post {
url: url.into(),
body: body.into(),
content_type: content_type.into(),
}
}
/// The endpoint URL (both verbs carry one) — the empty-is-red config check reads it.
pub fn url(&self) -> &str {
match self {
Source::Get { url } | Source::Post { url, .. } => url,
}
}
/// Fetch the source bytes through `fetch`, dispatching on the verb.
fn fetch(&self, fetch: &crate::daemon::Fetcher) -> Result<Vec<u8>> {
match self {
Source::Get { url } => fetch.get(url),
Source::Post {
url,
body,
content_type,
} => fetch.post(url, body, content_type),
}
}
}
/// The outcome of resolving a daemon's source for one tick: a ready request, or a
/// missing REQUIRED secret (a loud `Failed` fact, with the reason to surface — never a
/// silent green). A GET daemon with no secret is always `Ready`; a daemon that needs a key
/// it cannot get from the [`SecretStore`](crate::daemon::SecretStore) is `MissingSecret`.
pub enum SourceResolution {
Ready(Source),
MissingSecret(String),
}
/// Set the phase both in the shared run state and out through the ctx (jera + BIOS).
fn set(run: &Arc<Mutex<IcebergRun>>, ctx: &CollectorCtx, phase: Phase) {
run.lock().unwrap_or_else(|e| e.into_inner()).phase = phase.clone();
ctx.set_phase(&phase);
}
fn note_error(run: &Arc<Mutex<IcebergRun>>, reason: String) {
run.lock().unwrap_or_else(|e| e.into_inner()).error = Some(reason);
}
/// **One full reconcile pass for a warehouse-sink daemon** — the shared body of the
/// daemon loop, exposed so tests drive it deterministically. Fetches `url`
/// (off-reactor via `spawn_blocking` — the network I/O never blocks the loop), parses
/// with `parse` (also off-reactor — a multi-MB RDF dump is CPU-bound), and lands the
/// batch in `table` through [`CollectorCtx::warehouse`], reaching **Online** — or a
/// [`Phase::Failed`] fact (no source / fetch error / parse error / **empty-is-RED**).
async fn reconcile_iceberg(
run: &Arc<Mutex<IcebergRun>>,
ctx: &CollectorCtx,
src: &Source,
table: &str,
parse: fn(&[u8]) -> Result<RecordBatch>,
) -> Result<()> {
run.lock().unwrap_or_else(|e| e.into_inner()).table = table.to_string();
set(run, ctx, Phase::Offline);
let url = src.url();
// ── 0. The INCREMENTAL tick, when the source's own twin declares a delta. ────
//
// O2c. A daemon reaches its registry twin by name (possible only since the two
// rosters were made to spell every source the same way), and the twin is the half
// that owns the freshness declaration, the business key and the whole
// resolve-plan → land → advance-the-mark discipline. A twin that says it re-reads
// everything has nothing to offer here, so it keeps the full path below and this
// costs it nothing.
if !url.trim().is_empty() {
if let Some(twin) = crate::by_name(&ctx.name).filter(|c| !c.freshness().rereads_everything())
{
return reconcile_incremental(run, ctx, twin.as_ref(), &url, table).await;
}
}
if url.trim().is_empty() {
let reason = "no source URL configured (set `url` in Settings)".to_string();
ctx.log.line(&reason);
set(
run,
ctx,
Phase::Failed {
reason: reason.clone(),
},
);
note_error(run, reason);
return Ok(()); // a missing config is a FACT (red), not a panic
}
// ── 1. Fetch the source (off the async runtime — a blocking rustls GET/POST). ──
set(run, ctx, Phase::Downloading { pct: 0.0 });
let fetch = ctx.fetch.clone();
let src_owned = src.clone();
let bytes = match tokio::task::spawn_blocking(move || src_owned.fetch(&fetch)).await {
Ok(Ok(b)) => b,
Ok(Err(e)) => {
let reason = format!("fetch {url}: {e:#}");
ctx.log.line(&reason);
set(
run,
ctx,
Phase::Failed {
reason: reason.clone(),
},
);
note_error(run, reason);
return Ok(()); // a fetch failure is a fact the loop retries next tick
}
Err(e) => return Err(anyhow!("fetch task join: {e}")),
};
set(run, ctx, Phase::Downloading { pct: 1.0 });
ctx.log
.line(format!("fetched {} bytes from {url}", bytes.len()));
// ── 2. Parse → Arrow (off the async runtime — CPU-bound for large sources). ───
set(run, ctx, Phase::Indexing);
let batch = match tokio::task::spawn_blocking(move || parse(&bytes)).await {
Ok(Ok(b)) => b,
Ok(Err(e)) => {
let reason = format!("parse: {e:#}");
ctx.log.line(&reason);
set(
run,
ctx,
Phase::Failed {
reason: reason.clone(),
},
);
note_error(run, reason);
return Err(e);
}
Err(e) => return Err(anyhow!("parse task join: {e}")),
};
// ── 3. empty-is-RED — an enabled collector producing nothing is a failure. ────
let rows = batch.num_rows() as u64;
if rows == 0 {
let reason = format!("enabled but parsed 0 rows into `{table}` (empty-is-RED)");
ctx.log.line(&reason);
set(
run,
ctx,
Phase::Failed {
reason: reason.clone(),
},
);
note_error(run, reason);
return Ok(());
}
// ── 4. Land the batch in the warehouse → Online (off-reactor — a live Iceberg
// fast_append is blocking I/O, so it never stalls the daemon loop). ────────
let warehouse = ctx.warehouse.clone();
let table_owned = table.to_string();
let appended = tokio::task::spawn_blocking(move || warehouse.append(&table_owned, &batch))
.await
.map_err(|e| anyhow!("append task join: {e}"))?
.context("warehouse append")?;
{
let mut g = run.lock().unwrap_or_else(|e| e.into_inner());
g.rows_written = appended;
g.last_run_unix = now_unix();
g.error = None;
}
set(run, ctx, Phase::Online);
ctx.log.line(format!("ONLINE — {appended} rows → {table}"));
Ok(())
}
/// **One incremental reconcile pass, run through the daemon's registry TWIN.**
///
/// The daemon half owns the loop, the phase reporting and the sink; the registry half
/// owns the freshness declaration, the plan, the parse and the watermark. This routes
/// one tick through the second half so the daemon inherits everything
/// [`run_incremental`](crate::run_incremental) already guarantees — most of all that
/// **the mark advances only after the flush**, so a failure re-does the slice instead
/// of skipping it.
///
/// Three things it must get right, each of which was a way to be silently wrong:
///
/// * **The fixture reaches the twin.** `Collector::collect` builds its own HTTP core
/// from the cfg, so `ctx.fetch.canned()` is threaded into `CollectorCfg::canned`.
/// Without it a hermetic daemon test calls the live source (measured: a canned
/// polisen tick landed 500 rows — the live page size).
/// * **An explained empty is ONLINE.** A source that proved unchanged is a success and
/// the tick's whole purpose; `empty-is-RED` must fire only for an *unexplained* zero,
/// which is what `CollectStats::is_red` already distinguishes.
/// * **The rate cap is the source's own.** The twin carries it (Digitraffic 60/min ⇒
/// 1/s); the leaf `Fetcher` never had one. This path is therefore politer to the
/// source than the one it replaces, not merely fresher.
#[allow(clippy::too_many_arguments)]
async fn reconcile_incremental(
run: &Arc<Mutex<IcebergRun>>,
ctx: &CollectorCtx,
twin: &dyn crate::Collector,
url: &str,
table: &str,
) -> Result<()> {
set(run, ctx, Phase::Downloading { pct: 0.0 });
let cfg = {
let c = crate::CollectorCfg::new(url, table);
match ctx.fetch.canned() {
Some(canned) => c.with_canned(canned),
None => c,
}
};
set(run, ctx, Phase::Indexing);
let outcome = crate::run_incremental(twin, &cfg, ctx.warehouse.clone(), ctx.marks.as_ref()).await;
let inc = match outcome {
Ok(i) => i,
Err(e) => {
let reason = format!("incremental {url}: {e:#}");
ctx.log.line(&reason);
set(
run,
ctx,
Phase::Failed {
reason: reason.clone(),
},
);
note_error(run, reason);
// A fetch/parse/land failure is a FACT the loop retries next tick with
// backoff — the same contract the full path keeps.
return Ok(());
}
};
// empty-is-RED, but only for the UNEXPLAINED empty. `is_red` is the one writer of
// that distinction and `run_incremental` has already stamped RAGNARÖK on a zero
// whose collector could not say why.
if inc.stats.is_red() {
let reason = format!(
"enabled but landed 0 rows into `{table}` (empty-is-RED): {}",
inc.stats.reason.as_deref().unwrap_or("no reason given")
);
ctx.log.line(&reason);
set(
run,
ctx,
Phase::Failed {
reason: reason.clone(),
},
);
note_error(run, reason);
return Ok(());
}
{
let mut g = run.lock().unwrap_or_else(|e| e.into_inner());
g.rows_written = inc.stats.rows;
g.last_run_unix = now_unix();
g.error = None;
}
set(run, ctx, Phase::Online);
// The plan → outcome line, which is what makes an unchanged tick READABLE rather
// than an unexplained silence in the log.
ctx.log.line(format!("ONLINE — {}", inc.line()));
Ok(())
}
/// The generic never-ending daemon loop for a warehouse-sink collector: reconcile
/// toward ONLINE, then watch/refresh until [`CollectorCtx::shutdown`] is tripped.
/// **The wait after `failures` consecutive failed ticks.**
///
/// A daemon whose source is down retried on its ordinary cadence, forever. For a
/// query-window collector that is a request every five minutes against a server that is
/// already failing, from every korp in the fleet, for as long as the outage lasts —
/// which is the shape of a thundering herd, and it is the collector's own contribution
/// to keeping the source down.
///
/// So the interval doubles from [`BACKOFF_BASE_SECS`] and stops at [`BACKOFF_CAP_SECS`]
/// — one hour, because a source that has been down an hour deserves an hourly knock and
/// nothing sparser, or nobody notices when it comes back.
///
/// It NEVER polls faster than the ordinary cadence. That is what the `max` is for, and
/// it is not a detail: `AlwaysFull` sources wait a day between reads, so a cap of one
/// hour applied blindly would make a FAILING full-catalogue crawl hammer the source
/// twenty-four times harder than a healthy one. Backoff that speeds up on failure is
/// worse than no backoff at all.
///
/// `failures == 0` is a healthy tick and returns the ordinary cadence unchanged.
fn backoff_secs(refresh_secs: u64, failures: u32) -> u64 {
if failures == 0 {
return refresh_secs;
}
let grown = BACKOFF_BASE_SECS
.checked_shl(failures - 1)
.unwrap_or(BACKOFF_CAP_SECS)
.min(BACKOFF_CAP_SECS);
refresh_secs.max(grown)
}
/// Where the doubling starts. Small enough that a brief blip costs one short wait.
const BACKOFF_BASE_SECS: u64 = 30;
/// Where the doubling stops. An hour: sparse enough to stop being part of the outage,
/// frequent enough that recovery is noticed the same working hour it happens.
const BACKOFF_CAP_SECS: u64 = 3_600;
/// **The pre-flight gate: refuse at load, or not at all.**
///
/// The owner's ruling, 2026-09-06 — a collector whose config is not filled in is NOT
/// STARTABLE: it "refuses at load with a named reason and never half-starts, never
/// retries, never logs a nag".
///
/// Returns `true` when the daemon must NOT start. It writes the terminal
/// [`Phase::Refused`] and logs the reason **once**, then the caller returns without
/// entering the loop. There is no second call site and no second wording: the sentence
/// comes from [`Requirements::refusal`], which every other face of the collector also
/// uses.
///
/// The check reads the daemon's OWN declaration against the ctx's live secret store and
/// its live config — so a host that injects a vault-backed `SecretStore` is assessed on
/// what the vault actually answers, not on what the process environment happens to hold.
fn refuse_before_start(
run: &Arc<Mutex<IcebergRun>>,
ctx: &CollectorCtx,
requirements: &crate::requires::Requirements,
) -> bool {
let presence = crate::requires::StorePresence(ctx.secret.as_ref());
let Some(reason) = requirements.refusal(&presence, &ctx.config) else {
return false;
};
ctx.log.line(&reason);
set(run, ctx, Phase::Refused { reason: reason.clone() });
note_error(run, reason);
true
}
async fn daemon_run_loop(
run: Arc<Mutex<IcebergRun>>,
ctx: CollectorCtx,
requirements: crate::requires::Requirements,
mut provide_source: impl FnMut(&CollectorCtx) -> SourceResolution,
table: String,
parse: fn(&[u8]) -> Result<RecordBatch>,
refresh_secs: u64,
) {
// ── The gate. Nothing below this line runs for an unconfigured collector. ──────
if refuse_before_start(&run, &ctx, &requirements) {
return;
}
// Consecutive failed ticks, for the backoff. Reset by any tick that reaches a
// phase other than `Failed` — the PHASE is the verdict, not whether `reconcile`
// returned `Ok`, because reconcile answers `Ok` for every fact it records
// (no URL, fetch error, empty-is-RED) and reserves `Err` for a join failure.
// Reading the return value instead would have backed off on none of them.
let mut failures: u32 = 0;
loop {
// Re-resolve the source EACH tick: a source URL can change under a config edit,
// and a fetch failure is genuinely transient. What is NOT transient is a missing
// credential — see the `MissingSecret` arm.
match provide_source(&ctx) {
SourceResolution::Ready(src) => {
if let Err(e) = reconcile_iceberg(&run, &ctx, &src, &table, parse).await {
ctx.log
.line(format!("reconcile failed (will retry): {e:#}"));
}
}
// **A missing secret ends the daemon. It does not fail it.**
//
// This arm used to read: log the reason, `Phase::Failed`, fall through, back
// off, tick again — forever — on the argument that "a key absent now may
// appear later (an env var set, a vault lease granted)". That argument is
// what produced an hourly nag per unconfigured collector, and the owner ruled
// against it: never half-starts, never retries, never nags. A key that
// appears later is a RESTART, which the host already does when settings
// change; it is not something to poll a human for.
//
// Reaching here at all means the declaration missed something the resolver
// knows — `refuse_before_start` above catches every DECLARED requirement, so
// this is the belt to its braces. It is still terminal, and the
// `every_credential_gated_collector_names_its_variable` law test exists to
// keep this arm unreachable in practice.
SourceResolution::MissingSecret(reason) => {
ctx.log.line(&reason);
set(&run, &ctx, Phase::Refused { reason: reason.clone() });
note_error(&run, reason);
return;
}
}
if ctx.shutdown.is_stopped() {
break;
}
// The PHASE the tick reached is the verdict. `reconcile_iceberg` answers `Ok`
// for every fact it records — no URL, fetch error, parse error, empty-is-RED —
// and keeps `Err` for a join failure, so a backoff keyed on the return value
// would have backed off on almost nothing.
let failed = matches!(
run.lock().unwrap_or_else(|e| e.into_inner()).phase,
Phase::Failed { .. }
);
failures = if failed { failures.saturating_add(1) } else { 0 };
let wait = backoff_secs(refresh_secs, failures);
if failed && wait > refresh_secs {
ctx.log.line(format!(
"{failures} failed ticks in a row — backing off to {wait}s (from {refresh_secs}s)"
));
}
let sleep = std::time::Duration::from_secs(wait);
let step = std::time::Duration::from_millis(200);
let mut waited = std::time::Duration::ZERO;
while waited < sleep {
if ctx.shutdown.is_stopped() {
break;
}
tokio::time::sleep(step).await;
waited += step;
}
if ctx.shutdown.is_stopped() {
break;
}
}
ctx.log.line("daemon stopped");
}
/// The custom facet (matrix click-expand) shared shape for a warehouse-sink daemon.
fn iceberg_facet(name: &str, run: &Arc<Mutex<IcebergRun>>) -> serde_json::Value {
let s = run.lock().unwrap_or_else(|e| e.into_inner()).clone();
serde_json::json!({
"collector": name,
"sink": "iceberg",
"phase": s.phase.label(),
"state": s.phase,
"progress": s.phase.progress(),
"table": s.table,
"rows_written": s.rows_written,
"last_run_unix": s.last_run_unix,
"error": s.error,
})
}
fn cfg_str(config: &serde_json::Value, key: &str, default: &str) -> String {
config
.get(key)
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.unwrap_or(default)
.to_string()
}
/// **The default wait between two reads of `name`'s source.**
///
/// Read from the source's own declared [`Freshness`](crate::freshness::Freshness), via
/// the daemon's registry TWIN — the `Collector` that reads the same source through the
/// `collect()` side. Reaching that twin is possible only because the two rosters now
/// spell every source the same way (adc4bc3); while they drifted, this lookup would
/// have missed four collectors and silently handed them the AlwaysFull default.
///
/// A name with no twin has no declaration, and the honest reading of "nobody has
/// described this source" is [`Freshness::default()`] — it re-reads everything, so it
/// re-reads it daily.
fn default_refresh_for(name: &str) -> u64 {
crate::by_name(name)
.map(|c| c.freshness())
.unwrap_or_default()
.default_refresh_secs()
}
/// The wait between two reads: the operator's `refresh_secs` if they set one, else the
/// cadence the source's kind implies. It was a flat 3600 for every daemon in the roster
/// — see [`crate::freshness::Freshness::default_refresh_secs`] for what that cost.
fn refresh_secs(config: &serde_json::Value, name: &str) -> u64 {
config
.get("refresh_secs")
.and_then(|v| v.as_u64())
.unwrap_or_else(|| default_refresh_for(name))
}
// ═════════════════════════════════════════════════════════════════════════════
// The per-collector parse seams — each rides the crate's existing pure parser.
// ═════════════════════════════════════════════════════════════════════════════
/// DCAT-AP-SE RDF/XML → the `(dataset, distribution)` registry batch.
fn parse_dcat(bytes: &[u8]) -> Result<RecordBatch> {
let rows = crate::dcat::parse_dcat_rdfxml(bytes)?;
crate::dcat::registry_batch(&rows)
}
/// PxWeb / JSON-stat2 cube → a code-column-per-dimension + `value` batch.
fn parse_pxweb(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value = serde_json::from_slice(bytes).context("pxweb: body is not JSON")?;
crate::pxweb::parse_jsonstat2(&v)
}
/// OSM `boundary=administrative` features (a JSON array of `{id, tags}`, as an
/// Overpass/features export) → the `admin_units` batch (WKB borders + the division
/// ladder). The pure tag-filter [`crate::osm_admin::extract_admin_units`] drops any
/// non-boundary feature.
fn parse_osm_admin(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value =
serde_json::from_slice(bytes).context("osm-admin: body is not JSON")?;
let arr = v
.as_array()
.ok_or_else(|| anyhow!("osm-admin: expected a JSON array of features"))?;
let features: Vec<(i64, String, Option<Vec<u8>>)> = arr
.iter()
.filter_map(|f| {
let id = f.get("id")?.as_i64()?;
let tags = f.get("tags")?;
Some((id, tags.to_string(), None))
})
.collect();
let units = crate::osm_admin::extract_admin_units(features);
crate::osm_admin::admin_batch(&units)
}
/// Nord Pool day-ahead document (JSON) → the per-`(area, period)` spot-price batch.
/// The document's period × bidding-zone matrix is fanned out to one flat row per
/// `(area, period)` by [`crate::nordpool::parse_nordpool_json`].
fn parse_nordpool(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value =
serde_json::from_slice(bytes).context("nordpool: body is not JSON")?;
let prices = crate::nordpool::parse_nordpool_json(&v)?;
crate::nordpool::prices_batch(&prices)
}
/// SDMX-JSON message (Eurostat 2.1 `data.structures[0]` or the 1.0 `structure`
/// shape) → the `(dims…, value)` observation batch. The series/observation split is
/// flattened directly to a `RecordBatch` by [`crate::sdmx::parse_sdmx_json`], whose
/// column set (one Utf8 code column per dimension + a nullable Float64 `value`) is
/// message-driven — the daemon lands whatever the message declares.
fn parse_sdmx(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value = serde_json::from_slice(bytes).context("sdmx: body is not JSON")?;
crate::sdmx::parse_sdmx_json(&v)
}
/// USGS earthquake GeoJSON `FeatureCollection` → the per-feature quake batch. Each
/// locatable `Point` feature is flattened to a quake row (id, place, magnitude,
/// depth, lon/lat, epoch-ms time, event type, url) by
/// [`crate::usgs::parse_usgs_geojson`]; unlocatable / non-Point features are dropped.
fn parse_usgs(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value = serde_json::from_slice(bytes).context("usgs: body is not JSON")?;
let quakes = crate::usgs::parse_usgs_geojson(&v)?;
crate::usgs::quakes_batch(&quakes)
}
/// OpenAQ air-quality `results` envelope (JSON) → per-measurement rows. Each
/// located `results[]` reading (numeric `value` + `parameter` + a `coordinates`
/// point) is flattened to a measurement row (station, parameter, value, unit,
/// lon/lat, country/city, UTC time) by [`crate::openaq::parse_openaq_json`];
/// value-less / parameter-less / unlocatable readings contribute no row.
fn parse_openaq(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value = serde_json::from_slice(bytes).context("openaq: body is not JSON")?;
let rows = crate::openaq::parse_openaq_json(&v)?;
crate::openaq::measurements_batch(&rows)
}
/// ERDDAP self-describing `.json` `table` document → a typed Arrow batch. The
/// `columnTypes` drive the Arrow types (String→Utf8, double/float→Float64,
/// int/long/…→Int64) and the `rows` matrix is unpacked column-wise directly to a
/// `RecordBatch` by [`crate::erddap::parse_erddap_json`] — one parser for every
/// tabledap dataset on every ERDDAP server, so the daemon lands whatever the
/// document's `columnNames`/`columnTypes` declare.
fn parse_erddap(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value = serde_json::from_slice(bytes).context("erddap: body is not JSON")?;
crate::erddap::parse_erddap_json(&v)
}
/// OGC WFS `GetFeature` GML `FeatureCollection` (XML) → feature rows. Each
/// `wfs:member` / `gml:featureMember` is streamed to one row (feature_id,
/// feature_type, srs_name, a WKT `geometry`, and the remaining leaves as a
/// `properties` JSON string) by [`crate::wfs::parse_wfs_gml`]; a feature with no
/// gml geometry still lands (null geometry).
fn parse_wfs(bytes: &[u8]) -> Result<RecordBatch> {
let xml = std::str::from_utf8(bytes).context("wfs: body is not valid UTF-8 XML")?;
let features = crate::wfs::parse_wfs_gml(xml)?;
crate::wfs::features_batch(&features)
}
/// Energy-Charts public-power `{unix_seconds, production_types}` document (JSON) →
/// per-`(time, production_type)` power rows. The parallel `unix_seconds[]` time
/// axis is zipped against each `production_types[].data[]` array by index and
/// transposed to one flat `(time, production_type, power_mw)` row per numeric cell
/// by [`crate::energy_charts::parse_energy_charts_json`]; null-padded / ragged /
/// non-numeric cells contribute no row.
fn parse_energy_charts(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value =
serde_json::from_slice(bytes).context("energy-charts: body is not JSON")?;
let readings = crate::energy_charts::parse_energy_charts_json(&v)?;
crate::energy_charts::power_batch(&readings)
}
/// RSS 2.0 / Atom 1.0 syndication feed (XML) → per-item rows. Each RSS `<item>` /
/// Atom `<entry>` is flattened to a row (feed_title, id, title, link, summary,
/// author, published, categories) by [`crate::rss::parse_feed`] — dialect-agnostic
/// (the item boundary is `item`/`entry`) via a pure-Rust quick-xml parse; a
/// document with no item/entry elements contributes no row.
fn parse_rss(bytes: &[u8]) -> Result<RecordBatch> {
let xml = std::str::from_utf8(bytes).context("rss: body is not valid UTF-8 XML")?;
let items = crate::rss::parse_feed(xml)?;
crate::rss::feed_batch(&items)
}
/// ENTSO-E market-document XML (`GL_MarketDocument` load / `Publication_MarketDocument`)
/// → per-`Point` rows. Each `TimeSeries/Period/Point` is walked to a row (in_domain,
/// bidding_zone resolved from the EIC domain, resolution, period_start, position,
/// quantity) by [`crate::entsoe::parse_entsoe_xml`]; a Point missing position or
/// quantity contributes no row.
fn parse_entsoe(bytes: &[u8]) -> Result<RecordBatch> {
let xml = std::str::from_utf8(bytes).context("entsoe: body is not valid UTF-8 XML")?;
let points = crate::entsoe::parse_entsoe_xml(xml)?;
crate::entsoe::points_batch(&points)
}
/// ENTSO-E day-ahead `Publication_MarketDocument` (`A44`, XML) → per-`Point` spot-price
/// rows. Each `TimeSeries/Period/Point` is walked to a row (in_domain, bidding_zone,
/// currency, resolution, period_start, position, price) reading the `<price.amount>`
/// leaf by [`crate::entsoe_day_ahead::parse_entsoe_day_ahead_xml`]; a `<quantity>`
/// (load) Point contributes no price row.
fn parse_entsoe_day_ahead(bytes: &[u8]) -> Result<RecordBatch> {
let xml =
std::str::from_utf8(bytes).context("entsoe-day-ahead: body is not valid UTF-8 XML")?;
let prices = crate::entsoe_day_ahead::parse_entsoe_day_ahead_xml(xml)?;
crate::entsoe_day_ahead::prices_batch(&prices)
}
/// ENTSO-E generation-per-type `GL_MarketDocument` (`A75`, XML) → per-`Point` rows.
/// Each `TimeSeries/Period/Point` is walked to a row (in_domain, bidding_zone,
/// psr_type, psr_label, resolution, period_start, position, quantity) — also capturing
/// the `<MktPSRType><psrType>` production-type code (B16=Solar, B19=Wind Onshore, …) —
/// by [`crate::entsoe_generation::parse_entsoe_generation_xml`].
fn parse_entsoe_generation(bytes: &[u8]) -> Result<RecordBatch> {
let xml =
std::str::from_utf8(bytes).context("entsoe-generation: body is not valid UTF-8 XML")?;
let points = crate::entsoe_generation::parse_entsoe_generation_xml(xml)?;
crate::entsoe_generation::gen_points_batch(&points)
}
/// GTFS-Realtime `VehiclePositions` protobuf `FeedMessage` (binary bytes) → the
/// per-vehicle live-position batch. The hand-rolled protobuf reader
/// [`crate::gtfs_rt::parse_gtfs_rt`] walks the varint/length-delimited wire format
/// and keeps each entity carrying a `vehicle` with a locatable `Position` (lat AND
/// lon); a `trip_update`/`alert` entity or a position-less vehicle contributes no
/// row (a whole feed of them is empty-is-RED). This is the crate's one continuous
/// stream — the daemon loop re-fetches the `.pb` endpoint on its refresh interval.
fn parse_gtfs_rt_vehicles(bytes: &[u8]) -> Result<RecordBatch> {
let rows = crate::gtfs_rt::parse_gtfs_rt(bytes)?;
crate::gtfs_rt::vehicles_batch(&rows)
}
/// GTFS **static** feed (a ZIP archive of CSV `.txt` members) → the `stops.txt`
/// stop-point batch (the config-default table of the multi-file feed). The
/// pure-Rust ZIP reader + typed CSV parse [`crate::gtfs::parse_gtfs_zip`] picks the
/// `stops.txt` member (by basename, so a folder-nested feed still resolves) and
/// lands the spec columns [`GtfsCollector::stops`](crate::gtfs::GtfsCollector::stops)
/// declares (id, name, WGS84 lat/lon, zone, location_type, parent_station); a feed
/// with no `stops.txt` is a loud parse error, a `stops.txt` with no data rows is
/// empty-is-RED. Rides the `gtfs` feature (flate2 + csv), so the daemon is gated.
#[cfg(feature = "gtfs")]
fn parse_gtfs_stops(bytes: &[u8]) -> Result<RecordBatch> {
let stops = crate::gtfs::GtfsCollector::stops();
crate::gtfs::parse_gtfs_zip(bytes, stops.file(), stops.cols())
}
/// World Bank Open Data indicator response (JSON) → per-`(country, year, indicator)`
/// statistics rows. The response is the World Bank `[meta, data]` two-element
/// envelope, so the observation array lives at `records_pointer == "/1"`; each
/// observation is flattened to a row (indicator_id, indicator, country,
/// countryiso3code, date, value) by the crate's existing REST/JSON record mapper
/// [`crate::rest::parse_records`] driven by the [`crate::rest::RestCollector::worldbank`]
/// preset's field map — the same generic JSON-pointer parser the Ingest REST
/// collector rides, reused here as the daemon's parse seam.
fn parse_worldbank(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value =
serde_json::from_slice(bytes).context("worldbank: body is not JSON")?;
let c = crate::rest::RestCollector::worldbank();
crate::rest::parse_records(&v, c.records_pointer(), c.fields())
}
/// GBIF occurrence-search response (JSON) → per-occurrence biodiversity rows. The
/// `results[]` array is flattened to a row (gbif_key, scientific_name, kingdom,
/// lat, lon, country, basis_of_record, dataset_key, event_date) by
/// [`crate::rest::parse_records`] driven by the
/// [`crate::rest::RestCollector::gbif`] preset's field map. A record missing a
/// field contributes a null cell for it (the record still lands); an empty
/// `results[]` is empty-is-RED.
fn parse_gbif(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value = serde_json::from_slice(bytes).context("gbif: body is not JSON")?;
let c = crate::rest::RestCollector::gbif();
crate::rest::parse_records(&v, c.records_pointer(), c.fields())
}
/// EEA WISE water-quality response (JSON, the Discodata SQL API `results` envelope)
/// → per-`(site, determinand, year)` water-quality rows. The `results[]` array is
/// flattened to a row (monitoring_site, country, water_body_category, determinand,
/// mean_value, unit, year, lat, lon) by [`crate::rest::parse_records`] driven by the
/// [`crate::rest::RestCollector::eea_wise`] preset's field map.
fn parse_eea_wise(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value =
serde_json::from_slice(bytes).context("eea-wise: body is not JSON")?;
let c = crate::rest::RestCollector::eea_wise();
crate::rest::parse_records(&v, c.records_pointer(), c.fields())
}
/// **Polisen** incident-notice `events` (a top-level JSON array) → per-event rows.
/// Each event is flattened to a row (id, datetime, name, type, summary, url,
/// location_name, gps) by [`crate::rest::parse_records`] driven by the
/// [`crate::rest::RestCollector::polisen`] preset's field map (records at the document
/// root — the empty pointer). No auth, no personal data; an empty array is empty-is-RED.
fn parse_polisen(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value = serde_json::from_slice(bytes).context("polisen: body is not JSON")?;
let c = crate::rest::RestCollector::polisen();
crate::rest::parse_records(&v, c.records_pointer(), c.fields())
}
/// **SMHI** open weather observation STATIONS (a metobs parameter station-set, the
/// `station[]` geometry array) → per-station point rows. Each station is flattened to a
/// row (station_key, name, owner, lat, lon, height, active) by
/// [`crate::rest::parse_records`] driven by the
/// [`crate::rest::RestCollector::smhi_weather`] preset's field map (records at
/// `/station`). CC-BY 4.0, no key; a parameter with no stations is empty-is-RED.
fn parse_smhi_weather(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value = serde_json::from_slice(bytes).context("smhi: body is not JSON")?;
let c = crate::rest::RestCollector::smhi_weather();
crate::rest::parse_records(&v, c.records_pointer(), c.fields())
}
/// **Trafikverket** `TrainPosition` `data.json` response → the moving-map position
/// batch. The service answers a query POSTed as an XML `<REQUEST>` body (see the
/// [`TrafikverketDaemon`] below, which builds it), and the JSON reply carries the result
/// array at `/RESPONSE/RESULT/0/TrainPosition`. Each record is flattened to a row
/// (train_id, wgs84, timestamp, bearing, speed, status_active) by
/// [`crate::rest::parse_records`] driven by the
/// [`crate::trafikverket::TrafikverketCollector::train_positions`] field map; an empty
/// result array is empty-is-RED.
fn parse_trafikverket_trainpos(bytes: &[u8]) -> Result<RecordBatch> {
let v: serde_json::Value =
serde_json::from_slice(bytes).context("trafikverket: body is not JSON")?;
let c = crate::trafikverket::TrafikverketCollector::train_positions();
crate::rest::parse_records(&v, &c.records_pointer(), c.fields())
}
// ═════════════════════════════════════════════════════════════════════════════
// The daemons — thin structs over the shared skeleton.
// ═════════════════════════════════════════════════════════════════════════════
/// Generate a warehouse-sink daemon struct + its `Config`/`State`/`Collector` impls.
/// Each differs only in name, default source URL, default table, the config schema,
/// and which parse seam it rides — everything else is the shared skeleton above.
macro_rules! iceberg_daemon {
(
$daemon:ident, $name:literal, $default_url:literal, $default_table:literal,
title = $title:literal, desc = $desc:literal,
parse = $parse:path,
schema = $schema:expr $(,)?
) => {
#[doc = concat!("The `", $name, "` warehouse-sink collector daemon.")]
pub struct $daemon {
enabled: bool,
config: serde_json::Value,
run: Arc<Mutex<IcebergRun>>,
}
impl Default for $daemon {
fn default() -> Self {
Self::new()
}
}
impl $daemon {
#[doc = "The stable collector name (registry key · Settings key · matrix cell)."]
pub const NAME: &'static str = $name;
/// A disabled daemon with default config (the registry entry; korp overlays
/// the persisted `enabled` + custom config before starting it).
pub fn new() -> Self {
Self { enabled: false, config: Self::default_config(), run: Arc::new(Mutex::new(IcebergRun::default())) }
}
/// A daemon with an explicit `enabled` + custom config (what korp builds
/// from the persisted settings before spawning [`Collector::run`]).
pub fn with_config(enabled: bool, config: serde_json::Value) -> Self {
Self { enabled, config, run: Arc::new(Mutex::new(IcebergRun::default())) }
}
/// The Settings defaults for this collector.
pub fn default_config() -> serde_json::Value {
serde_json::json!({ "url": $default_url, "table": $default_table })
}
/// Read the live run snapshot.
pub fn snapshot(&self) -> IcebergRun {
self.run.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
/// The configured source URL (defaults to the collector's canonical endpoint).
pub fn url(&self) -> String {
cfg_str(&self.config, "url", $default_url)
}
/// The configured Iceberg table (defaults to the collector's canonical table).
pub fn table(&self) -> String {
cfg_str(&self.config, "table", $default_table)
}
/// **One full reconcile pass** (fetch → parse → warehouse → Online), exposed
/// so tests drive it deterministically without the never-ending wrapper.
pub async fn reconcile_once(&self, ctx: &CollectorCtx) -> Result<()> {
reconcile_iceberg(&self.run, ctx, &Source::get(self.url()), &self.table(), $parse)
.await
}
}
impl CollectorConfig for $daemon {
fn name(&self) -> &str {
Self::NAME
}
fn enabled(&self) -> bool {
self.enabled
}
fn schema(&self) -> ConfigSchema {
$schema
}
fn title(&self) -> String {
$title.to_string()
}
fn description(&self) -> String {
$desc.to_string()
}
fn defaults(&self) -> serde_json::Value {
Self::default_config()
}
}
impl CollectorState for $daemon {
fn phase(&self) -> Phase {
self.snapshot().phase
}
fn facet(&self) -> serde_json::Value {
iceberg_facet(Self::NAME, &self.run)
}
}
#[async_trait::async_trait]
impl Collector for $daemon {
async fn run(self: Arc<Self>, ctx: CollectorCtx) {
let (url, table, secs) = (self.url(), self.table(), refresh_secs(&self.config, Self::NAME));
// A GET daemon needs no secret — its source is always Ready.
let requirements = CollectorConfig::requirements(&*self);
daemon_run_loop(
self.run.clone(),
ctx,
requirements,
move |_ctx| SourceResolution::Ready(Source::get(url.clone())),
table,
$parse,
secs,
)
.await;
}
}
};
}
iceberg_daemon!(
OsmAdminDaemon,
"osm-admin-boundaries",
"", // Overpass admin-boundary query; wired per-deployment (empty ⇒ red until set)
"admin_units",
title = "🏛 osm-admin-boundaries",
desc = "Fetch OSM boundary=administrative relations and land the admin-unit ladder \
(län/kommun WKB borders) in an Iceberg table.",
parse = parse_osm_admin,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"Source",
"Overpass/features JSON of boundary=administrative relations"
),
ConfigField::text(
"admin_levels",
"Admin levels",
"e.g. 4,7 (län + kommun) — advisory"
),
ConfigField::text("table", "Iceberg table", "admin_units"),
]),
);
iceberg_daemon!(
DcatDaemon,
"dcat-ap-se",
"https://admin.dataportal.se/all.rdf",
"dcat_registry",
title = "📚 dcat-ap-se",
desc = "Fetch the DCAT-AP-SE catalogue (dataportal.se RDF/XML) and land the \
(dataset, distribution) registry in an Iceberg table.",
parse = parse_dcat,
schema = ConfigSchema::new(vec![
ConfigField::text("url", "Catalogue", "dataportal.se RDF/XML dump URL"),
ConfigField::text("mode", "Mode", "rdf | sparql"),
ConfigField::text("table", "Iceberg table", "dcat_registry"),
]),
);
iceberg_daemon!(
PxWebDaemon,
"pxweb-jsonstat2",
"https://api.scb.se/OV0104/v1/doris/sv/ssd",
"pxweb_cube",
title = "📊 pxweb-jsonstat2",
desc = "Fetch a PxWeb / JSON-stat2 statistics cube (SCB or any PxWeb 2.0 agency) \
and land its cells (code-column-per-dimension + value) in an Iceberg table.",
parse = parse_pxweb,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"PxWeb table",
"SCB/agency PxWeb 2.0 JSON-stat2 endpoint"
),
ConfigField::text(
"selection",
"Selection",
"optional JSON cell selection (blank ⇒ GET the cube)"
),
ConfigField::text("table", "Iceberg table", "pxweb_cube"),
]),
);
iceberg_daemon!(
NordPoolDaemon,
"nordpool",
"https://dataportal-api.nordpoolgroup.com/api/DayAheadPrices",
"nordpool_prices",
title = "⚡ nordpool",
desc = "Fetch the Nord Pool day-ahead spot-price document and land the period×zone \
matrix flattened to per-(area, period) price rows (SE1–SE4 / NO / DK / FI \
bidding zones) in an Iceberg table.",
parse = parse_nordpool,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"Day-ahead endpoint",
"Nord Pool data-portal DayAheadPrices URL"
),
ConfigField::text("areas", "Bidding zones", "e.g. SE1,SE2,SE3,SE4 — advisory"),
ConfigField::text("table", "Iceberg table", "nordpool_prices"),
]),
);
iceberg_daemon!(
SdmxDaemon,
"eurostat-sdmx",
"https://ec.europa.eu/eurostat/api/dissemination/sdmx/2.1/data",
"eurostat_sdmx",
title = "🇪🇺 eurostat-sdmx",
desc = "Fetch an SDMX-JSON 2.1 statistics message (Eurostat or any SDMX provider) \
and land its series/observation split flattened to (dims…, value) rows in \
an Iceberg table.",
parse = parse_sdmx,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"SDMX data query",
"SDMX 2.1 REST data endpoint (format=jsondata)"
),
ConfigField::text(
"dims",
"Dimensions",
"optional declared dimension ids — advisory"
),
ConfigField::text("table", "Iceberg table", "eurostat_sdmx"),
]),
);
iceberg_daemon!(
UsgsDaemon,
"usgs-earthquakes",
"https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_hour.geojson",
"usgs_quakes",
title = "🌎 usgs-earthquakes",
desc = "Fetch the USGS real-time earthquake GeoJSON FeatureCollection and land each \
locatable Point feature as a quake row (id, place, magnitude, depth, lon/lat, \
epoch-ms time, event type, url) in an Iceberg table.",
parse = parse_usgs,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"GeoJSON feed",
"USGS summary FeatureCollection URL (e.g. all_hour.geojson)"
),
ConfigField::text("min_magnitude", "Min magnitude", "advisory magnitude floor"),
ConfigField::text("table", "Iceberg table", "usgs_quakes"),
]),
);
iceberg_daemon!(
OpenAqDaemon,
"openaq-air-quality",
"https://api.openaq.org/v3/measurements",
"openaq_measurements",
title = "🌬 openaq-air-quality",
desc = "Fetch the OpenAQ air-quality measurements API and land each located \
reading (station, parameter, value, unit, lon/lat, country/city, UTC time) \
as a measurement row in an Iceberg table (CC BY 4.0; needs an X-API-Key).",
parse = parse_openaq,
schema = ConfigSchema::new(vec![
ConfigField::text("url", "Measurements endpoint", "OpenAQ v3 measurements URL"),
ConfigField::text("parameters", "Parameters", "e.g. pm25,no2,o3 — advisory"),
ConfigField::text("table", "Iceberg table", "openaq_measurements"),
]),
);
iceberg_daemon!(
ErddapDaemon,
"erddap-emodnet-physics",
"https://erddap.emodnet-physics.eu/erddap/tabledap/EP_ERD_INT_TZTM_AL_TS_NRT.json",
"erddap_emodnet_physics",
title = "🌊 erddap-emodnet-physics",
desc = "Fetch an ERDDAP self-describing .json table (EMODnet Physics or any ERDDAP \
tabledap/griddap server) and land its columnNames/columnTypes-typed rows \
(String→Utf8, double/float→Float64, int→Int64) in an Iceberg table.",
parse = parse_erddap,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"ERDDAP .json request",
"…/tabledap/<dataset>.json?<query>"
),
ConfigField::text("dataset_id", "Dataset id", "ERDDAP dataset id — advisory"),
ConfigField::text("table", "Iceberg table", "erddap_emodnet_physics"),
]),
);
iceberg_daemon!(
WfsDaemon,
"wfs",
"https://example.wfs/geoserver/ows?service=WFS&version=2.0.0&request=GetFeature",
"wfs_features",
title = "🗺 wfs",
desc = "Fetch an OGC WFS GetFeature GML FeatureCollection (Lantmäteriet / SGU / \
SMHI geodata) and land each member as a feature row (WKT geometry + \
properties JSON) in an Iceberg table via a pure-Rust GML parse.",
parse = parse_wfs,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"GetFeature request",
"WFS GetFeature URL (service/version/typeNames)"
),
ConfigField::text("type_names", "Type names", "WFS typeNames — advisory"),
ConfigField::text("table", "Iceberg table", "wfs_features"),
]),
);
// WAVE 2, batch 3 — energy-charts / rss-atom / the ENTSO-E trio adopted to the
// warehouse-sink pattern (all pure `parsers`, so they register under `daemon` alone).
iceberg_daemon!(
EnergyChartsDaemon,
"energy-charts-power",
"https://api.energy-charts.info/public_power?country=de",
"energy_charts_power",
title = "⚡ energy-charts-power",
desc = "Fetch the Energy-Charts (Fraunhofer ISE) public-power document and land the \
parallel unix_seconds[] time axis zipped against each production_types[].data[] \
into per-(time, type) generation-mix power rows (MW) in an Iceberg table \
(Europe-wide via ?country=; CC-BY-4.0).",
parse = parse_energy_charts,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"Public-power endpoint",
"Energy-Charts public_power URL (?country=…)"
),
ConfigField::text(
"country",
"Country",
"e.g. de, se, eu — advisory (encode in the URL)"
),
ConfigField::text("table", "Iceberg table", "energy_charts_power"),
]),
);
iceberg_daemon!(
RssDaemon,
"rss-atom",
"https://example.se/feed.xml",
"feed_items",
title = "📰 rss-atom",
desc = "Fetch an RSS 2.0 / Atom 1.0 syndication feed (Krisinformation.se alerts, SMHI \
warnings, agency press wires) and land each item/entry as a row (feed_title, id, \
title, link, summary, author, published, categories) in an Iceberg table via a \
pure-Rust quick-xml parse.",
parse = parse_rss,
schema = ConfigSchema::new(vec![
ConfigField::text("url", "Feed URL", "RSS 2.0 or Atom 1.0 feed document URL"),
ConfigField::text(
"dialect",
"Dialect",
"rss | atom — advisory (auto-detected)"
),
ConfigField::text("table", "Iceberg table", "feed_items"),
]),
);
iceberg_daemon!(
EntsoeDaemon,
"entsoe",
"https://web-api.tp.entsoe.eu/api",
"entsoe_prices",
title = "🔌 entsoe",
desc = "Fetch an ENTSO-E Transparency Platform market document (EU electricity \
generation/load) and land each TimeSeries/Period/Point as a row (in_domain, \
bidding_zone, resolution, period_start, position, quantity) in an Iceberg table; \
SE1–SE4 via EIC domain codes (free token in the URL).",
parse = parse_entsoe,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"API request",
"ENTSO-E web-api URL incl. securityToken + documentType query"
),
ConfigField::text(
"in_domain",
"In domain",
"bidding zone / EIC domain — advisory (encode in the URL)"
),
ConfigField::text("table", "Iceberg table", "entsoe_prices"),
]),
);
iceberg_daemon!(
EntsoeDayAheadDaemon,
"entsoe_day_ahead",
"https://web-api.tp.entsoe.eu/api",
"entsoe_day_ahead_prices",
title = "💶 entsoe_day_ahead",
desc = "Fetch an ENTSO-E day-ahead prices document (documentType A44) and land each \
Point as a spot-price row (in_domain, bidding_zone, currency, resolution, \
period_start, position, price via price.amount) in an Iceberg table; the open \
(free-token) energy-market price feed, SE1–SE4 via EIC domain codes.",
parse = parse_entsoe_day_ahead,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"A44 request",
"ENTSO-E web-api URL incl. securityToken + documentType=A44"
),
ConfigField::text(
"in_domain",
"In domain",
"bidding zone / EIC domain — advisory (encode in the URL)"
),
ConfigField::text("table", "Iceberg table", "entsoe_day_ahead_prices"),
]),
);
iceberg_daemon!(
EntsoeGenerationDaemon,
"entsoe_generation",
"https://web-api.tp.entsoe.eu/api",
"entsoe_generation",
title = "🏭 entsoe_generation",
desc = "Fetch an ENTSO-E actual-generation-per-type document (documentType A75) and land \
each Point as a row (in_domain, bidding_zone, psr_type, psr_label, resolution, \
period_start, position, quantity) in an Iceberg table; the MktPSRType/psrType \
production type (B16=Solar, B19=Wind Onshore, …) is labelled, SE1–SE4 via EIC codes.",
parse = parse_entsoe_generation,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"A75 request",
"ENTSO-E web-api URL incl. securityToken + documentType=A75"
),
ConfigField::text(
"in_domain",
"In domain",
"bidding zone / EIC domain — advisory (encode in the URL)"
),
ConfigField::text("table", "Iceberg table", "entsoe_generation"),
]),
);
// WAVE 2, batch 4 — GTFS-Realtime (vehicles) + GTFS static (stops) adopted to the
// warehouse-sink pattern. gtfs-rt-vehicles rides the pure hand-rolled protobuf
// reader (no new dep), so it registers under `daemon` alone; gtfs-stops rides the
// pure-Rust ZIP + CSV reader (feature `gtfs` → flate2 + csv), so it is gated behind
// `gtfs` exactly as osm-cut is gated behind `osm-daemon`.
iceberg_daemon!(
GtfsRtVehiclesDaemon,
"gtfs-rt-vehicles",
"", // per-agency GTFS-RT `.pb` endpoint; wired per-deployment (empty ⇒ red until set)
"gtfs_rt_vehicles",
title = "🚌 gtfs-rt-vehicles",
desc = "Fetch a GTFS-Realtime VehiclePositions feed (a protobuf FeedMessage — the \
crate's one continuous stream, re-fetched each refresh tick) and land each \
locatable vehicle as a live-position row (entity_id, vehicle_id, trip_id, \
route_id, lat/lon, bearing, speed, timestamp, stop_id) in an Iceberg table \
via a hand-rolled protobuf reader (no prost dep); a trip_update/alert or a \
position-less vehicle contributes no row.",
parse = parse_gtfs_rt_vehicles,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"VehiclePositions .pb",
"GTFS-RT VehiclePositions protobuf endpoint (per-agency)"
),
ConfigField::text(
"refresh_secs",
"Refresh seconds",
"how often to re-fetch the live feed (default 3600) — advisory"
),
ConfigField::text("table", "Iceberg table", "gtfs_rt_vehicles"),
]),
);
#[cfg(feature = "gtfs")]
iceberg_daemon!(
GtfsStopsDaemon,
"gtfs-stops",
"", // per-agency GTFS static feed `.zip`; wired per-deployment (empty ⇒ red until set)
"gtfs_stops",
title = "🚏 gtfs-stops",
desc = "Fetch a GTFS static transit feed (a ZIP of CSV `.txt` members) and land the \
`stops.txt` member — the config-default table of the multi-file feed — as \
typed stop-point rows (id, name, WGS84 lat/lon, zone, location_type, \
parent_station) in an Iceberg table via a pure-Rust ZIP reader + csv parse; \
the member is picked by basename so a folder-nested feed still resolves.",
parse = parse_gtfs_stops,
schema = ConfigSchema::new(vec![
ConfigField::text("url", "Feed .zip", "GTFS static feed ZIP URL (per-agency)"),
ConfigField::text(
"file",
"Member",
"the landed GTFS member (stops.txt) — advisory"
),
ConfigField::text("table", "Iceberg table", "gtfs_stops"),
]),
);
// WAVE 2, batch 5 — the World Bank / GBIF / EEA-WISE open-data collectors adopted to
// the warehouse-sink pattern. Each rides the crate's existing generic REST/JSON record
// mapper (`rest::parse_records` + the matching `RestCollector` preset's field map), so
// they register under `daemon` alone (pure `parsers`, no new dep) exactly as the other
// JSON collectors do.
iceberg_daemon!(
WorldBankDaemon,
"worldbank-indicators",
"https://api.worldbank.org/v2/country/all/indicator/SP.POP.TOTL?format=json&per_page=1000",
"worldbank_indicators",
title = "🏦 worldbank-indicators",
desc = "Fetch a World Bank Open Data indicator series (the [meta, data] envelope, \
observations at /1) and land each observation as a per-(country, year, \
indicator) statistics row (indicator_id, indicator, country, countryiso3code, \
date, value) in an Iceberg table (CC-BY 4.0; public, no key).",
parse = parse_worldbank,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"Indicator query",
"World Bank v2 indicator URL (format=json)"
),
ConfigField::text(
"indicator",
"Indicator",
"e.g. SP.POP.TOTL — advisory (encode in the URL)"
),
ConfigField::text("table", "Iceberg table", "worldbank_indicators"),
]),
);
iceberg_daemon!(
GbifDaemon,
"gbif-occurrences",
"https://api.gbif.org/v1/occurrence/search?hasCoordinate=true&limit=300",
"gbif_occurrences",
title = "🦋 gbif-occurrences",
desc = "Fetch a GBIF occurrence-search page (results[] envelope) and land each \
occurrence as a biodiversity point row (gbif_key, scientific_name, kingdom, \
lat/lon, country, basis_of_record, dataset_key, event_date) in an Iceberg \
table (Global Biodiversity Information Facility; CC-BY/CC0, public, no key).",
parse = parse_gbif,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"Occurrence search",
"GBIF v1 occurrence/search URL (?country=&limit=)"
),
ConfigField::text(
"country",
"Country",
"e.g. SE — advisory (encode in the URL)"
),
ConfigField::text("table", "Iceberg table", "gbif_occurrences"),
]),
);
iceberg_daemon!(
EeaWiseDaemon,
"eea-wise-water",
"https://discodata.eea.europa.eu/sql?query=SELECT%20TOP%201000%20*%20FROM%20%5BWISE_SOE%5D.%5Bv2r1%5D.%5BWaterbase_S_WISE6_AggregatedData%5D",
"eea_wise",
title = "💧 eea-wise-water",
desc = "Fetch an EEA WISE water-quality query (the Discodata SQL API results[] \
envelope) and land each aggregated reading as a per-(site, determinand, year) \
row (monitoring_site, country, water_body_category, determinand, mean_value, \
unit, year, lat/lon) in an Iceberg table (EEA / Copernicus; CC-BY 4.0).",
parse = parse_eea_wise,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"Discodata SQL",
"EEA Discodata /sql?query=<SELECT …> URL"
),
ConfigField::text(
"determinand",
"Determinand",
"e.g. Nitrate — advisory (encode in the SQL)"
),
ConfigField::text("table", "Iceberg table", "eea_wise"),
]),
);
// WAVE 2, batch 6 — the Swedish real-time public-sector sources adopted to the
// warehouse-sink pattern. `polisen-events` and `smhi-weather` are plain-GET JSON, so
// they ride the macro exactly like the other JSON collectors; `trafikverket-trainpos`
// POSTs an XML query (with its authentication key in the body) and so is hand-written
// over the SAME shared `reconcile_iceberg`/`daemon_run_loop` — only its `Source` differs.
iceberg_daemon!(
PolisenDaemon,
"polisen-events",
"https://polisen.se/api/events",
"polisen_events",
title = "🚓 polisen-events",
desc = "Fetch Polisen incident notices (the events API — a top-level JSON array, no \
auth, no personal data) and land each event as a row (id, datetime, name, \
type, summary, url, location_name, gps) in an Iceberg table.",
parse = parse_polisen,
schema = ConfigSchema::new(vec![
ConfigField::text("url", "Events API", "https://polisen.se/api/events (optionally ?locationname=&type=)"),
ConfigField::text(
"region",
"Region",
"e.g. Stockholm — advisory (encode as ?locationname= in the URL)"
),
ConfigField::text("table", "Iceberg table", "polisen_events"),
]),
);
iceberg_daemon!(
SmhiWeatherDaemon,
"smhi-weather",
"https://opendata-download-metobs.smhi.se/api/version/latest/parameter/1.json",
"smhi_weather",
title = "🌦 smhi-weather",
desc = "Fetch an SMHI open weather observation STATION set (a metobs parameter \
station-set JSON, the `station[]` geometry array; CC-BY 4.0, no key) and land \
each station as a point row (station_key, name, owner, lat/lon, height, active) \
in an Iceberg table — the geometry the metobs value series joins back to.",
parse = parse_smhi_weather,
schema = ConfigSchema::new(vec![
ConfigField::text(
"url",
"Parameter station-set",
"SMHI metobs `/parameter/<n>.json` URL (n = 1 air temperature, …)"
),
ConfigField::text(
"parameter",
"Parameter",
"e.g. 1 (air temperature) — advisory (encode in the URL)"
),
ConfigField::text("table", "Iceberg table", "smhi_weather"),
]),
);
// ─────────────────────────────────────────────────────────────────────────────
// Trafikverket — the one POST-body daemon. Hand-written (not the GET macro) because
// its `Source` is a POST carrying an XML `<REQUEST>` with the authentication key, but
// it reuses the SAME `reconcile_iceberg`/`daemon_run_loop`/`iceberg_facet` as every
// other warehouse-sink daemon: only the request shape is bespoke, not the state machine.
// ─────────────────────────────────────────────────────────────────────────────
/// The `trafikverket-trainpos` warehouse-sink collector daemon — real-time train
/// positions POSTed as an XML query to the Trafikinfo API. The authentication key lives
/// in the request BODY (`<LOGIN authenticationkey=…>`), read from config `key` and
/// falling back to the env var named by `key_env` (default `TRAFIKVERKET_API_KEY`) — a
/// key is never hardcoded and, absent from config, never persisted.
pub struct TrafikverketDaemon {
enabled: bool,
config: serde_json::Value,
run: Arc<Mutex<IcebergRun>>,
}
impl Default for TrafikverketDaemon {
fn default() -> Self {
Self::new()
}
}
impl TrafikverketDaemon {
/// The stable collector name (registry key · Settings key · matrix cell).
pub const NAME: &'static str = "trafikverket-trainpos";
const DEFAULT_URL: &'static str = "https://api.trafikinfo.trafikverket.se/v2/data.json";
const DEFAULT_TABLE: &'static str = "trafikverket_trainpos";
const DEFAULT_OBJECTTYPE: &'static str = "TrainPosition";
/// A disabled daemon with default config (the registry entry; korp overlays the
/// persisted `enabled` + custom config before starting it).
pub fn new() -> Self {
Self {
enabled: false,
config: Self::default_config(),
run: Arc::new(Mutex::new(IcebergRun::default())),
}
}
/// A daemon with an explicit `enabled` + custom config.
pub fn with_config(enabled: bool, config: serde_json::Value) -> Self {
Self {
enabled,
config,
run: Arc::new(Mutex::new(IcebergRun::default())),
}
}
/// The Settings defaults for this collector. `key` is empty by default — a secret is
/// never seeded — and resolves from the `key_env` environment variable at run time.
pub fn default_config() -> serde_json::Value {
serde_json::json!({
"url": Self::DEFAULT_URL,
"table": Self::DEFAULT_TABLE,
"objecttype": Self::DEFAULT_OBJECTTYPE,
"key": "",
"key_env": "TRAFIKVERKET_API_KEY",
"query": "",
})
}
/// Read the live run snapshot.
pub fn snapshot(&self) -> IcebergRun {
self.run.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
/// The configured API endpoint.
pub fn url(&self) -> String {
cfg_str(&self.config, "url", Self::DEFAULT_URL)
}
/// The configured Iceberg table.
pub fn table(&self) -> String {
cfg_str(&self.config, "table", Self::DEFAULT_TABLE)
}
/// The configured objecttype (the result array selected in the reply).
fn objecttype(&self) -> String {
cfg_str(&self.config, "objecttype", Self::DEFAULT_OBJECTTYPE)
}
/// The name of the secret this daemon needs — the env-var name / vault-path leaf the
/// [`SecretStore`](crate::daemon::SecretStore) is asked for (config `key_env`, default
/// `TRAFIKVERKET_API_KEY`).
pub fn secret_name(&self) -> String {
cfg_str(&self.config, "key_env", "TRAFIKVERKET_API_KEY")
}
/// Resolve the authentication key **through the injected [`SecretStore`]** (design: no
/// daemon calls `std::env::var` directly, so a vault can be swapped in later). An inline
/// config `key` literal is an explicit Settings override that wins — matching the old
/// config-then-env order — and is otherwise deferred to the store (env today via
/// [`EnvSecretStore`](crate::daemon::EnvSecretStore), vault later). `None` (no override
/// AND the store has nothing) is the loud-`Failed` signal, never a silent empty.
fn resolve_key(
&self,
store: &dyn crate::daemon::SecretStore,
) -> Option<crate::daemon::SecretString> {
if let Some(k) = self
.config
.get("key")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
{
return Some(crate::daemon::SecretString::new(k));
}
store.secret(&self.secret_name())
}
/// The `<QUERY>` fragment — config `query` when set, else a minimal default.
fn query_xml(&self) -> String {
self.config
.get("query")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.map(|s| s.to_string())
.unwrap_or_else(|| {
format!(
"<QUERY objecttype=\"{}\" schemaversion=\"1.0\"/>",
self.objecttype()
)
})
}
/// Resolve the POST source for one tick: the XML `<REQUEST>` body (key from the store +
/// `<QUERY>`) to the endpoint, or [`SourceResolution::MissingSecret`] when the store has
/// no key. The `SecretString` is exposed ONLY here, into the request body, and dropped
/// (zeroized) at the end of the call — it is never logged or stored.
fn resolve_source(&self, store: &dyn crate::daemon::SecretStore) -> SourceResolution {
match self.resolve_key(store) {
Some(key) if !key.is_empty() => {
let body = crate::trafikverket::build_request(key.expose(), &self.query_xml());
SourceResolution::Ready(Source::post(self.url(), body, "text/xml"))
}
_ => SourceResolution::MissingSecret(format!(
"no Trafikverket key: set config `key`, the `{}` env var, or provide a \
SecretStore that resolves it (missing secret is RED, never a silent green)",
self.secret_name()
)),
}
}
/// **One full reconcile pass** (POST XML → parse → warehouse → Online), exposed so
/// tests drive it deterministically without the never-ending wrapper. A missing secret
/// short-circuits to a loud `Failed` — the daemon never fetches without its key.
pub async fn reconcile_once(&self, ctx: &CollectorCtx) -> Result<()> {
match self.resolve_source(ctx.secret.as_ref()) {
SourceResolution::Ready(src) => {
reconcile_iceberg(
&self.run,
ctx,
&src,
&self.table(),
parse_trafikverket_trainpos,
)
.await
}
SourceResolution::MissingSecret(reason) => {
ctx.log.line(&reason);
set(&self.run, ctx, Phase::Failed { reason: reason.clone() });
note_error(&self.run, reason);
Ok(()) // a missing secret is a FACT (red), not a panic
}
}
}
}
impl CollectorConfig for TrafikverketDaemon {
fn name(&self) -> &str {
Self::NAME
}
fn enabled(&self) -> bool {
self.enabled
}
fn schema(&self) -> ConfigSchema {
ConfigSchema::new(vec![
ConfigField::text("url", "Trafikinfo API", Self::DEFAULT_URL),
ConfigField::text(
"objecttype",
"Object type",
"TrainPosition (default) — the result array selected in the reply",
),
ConfigField::text(
"key",
"API key",
"Trafikverket authentication key (empty ⇒ resolved through the injected SecretStore: the key_env variable today, a vault later)",
),
ConfigField::text(
"key_env",
"Key env var",
"environment variable holding the key (default TRAFIKVERKET_API_KEY)",
),
ConfigField::text(
"query",
"Query XML",
"optional <QUERY …>…</QUERY> fragment (default: a minimal query for the objecttype)",
),
ConfigField::text("table", "Iceberg table", Self::DEFAULT_TABLE),
])
}
fn title(&self) -> String {
"🚆 trafikverket-trainpos".to_string()
}
fn description(&self) -> String {
"POST an XML query (authentication key in the body) to the Trafikverket \
Trafikinfo API and land the real-time TrainPosition reply as a moving-map \
position table (train_id, wgs84, timestamp, bearing, speed) in an Iceberg table."
.to_string()
}
fn defaults(&self) -> serde_json::Value {
Self::default_config()
}
}
impl CollectorState for TrafikverketDaemon {
fn phase(&self) -> Phase {
self.snapshot().phase
}
fn facet(&self) -> serde_json::Value {
iceberg_facet(Self::NAME, &self.run)
}
}
#[async_trait::async_trait]
impl Collector for TrafikverketDaemon {
async fn run(self: Arc<Self>, ctx: CollectorCtx) {
let (table, secs) = (self.table(), refresh_secs(&self.config, Self::NAME));
let me = self.clone();
// Re-resolve the secret through the ctx's SecretStore EACH tick: a key absent now
// (no env, vault not yet leased) may appear later, so it is a `Failed` this tick,
// not a dead daemon.
let requirements = CollectorConfig::requirements(&*self);
daemon_run_loop(
self.run.clone(),
ctx,
requirements,
move |ctx| me.resolve_source(ctx.secret.as_ref()),
table,
parse_trafikverket_trainpos,
secs,
)
.await;
}
}
#[cfg(test)]
mod tests {
use super::*;
/// **A failing daemon backs OFF, and never speeds up.**
///
/// Before this, a daemon whose source was down retried on its ordinary cadence
/// forever: a query-window collector is one request every five minutes against a
/// server that is already failing, from every korp in the fleet, for the length of
/// the outage. That is a thundering herd, and it is the collector's own
/// contribution to keeping the source down.
#[test]
fn a_failing_daemon_backs_off_and_a_healthy_one_does_not() {
const QUERY_WINDOW: u64 = 300;
// A healthy tick waits its ordinary cadence — backoff must be invisible until
// something actually fails.
assert_eq!(backoff_secs(QUERY_WINDOW, 0), QUERY_WINDOW);
// …and the wait then RISES, strictly, until it stops at the cap. Asserted as a
// shape rather than a table of numbers, so the constants can be retuned without
// rewriting the test — but the shape itself is the contract.
let walk: Vec<u64> = (0..12).map(|n| backoff_secs(QUERY_WINDOW, n)).collect();
assert!(
walk.windows(2).all(|w| w[1] >= w[0]),
"the wait must never DROP as failures accumulate: {walk:?}",
);
assert!(
walk.last().copied().unwrap_or(0) == BACKOFF_CAP_SECS,
"after 11 consecutive failures the wait must have reached the cap: {walk:?}",
);
assert!(
walk.iter().any(|w| *w > QUERY_WINDOW),
"the wait never rose above the ordinary cadence — that is not a backoff: {walk:?}",
);
// A cap that is never reached is a runaway; one reached immediately is not a
// backoff. Both are caught by requiring the rise to take more than one step.
let first_rise = walk.iter().position(|w| *w > QUERY_WINDOW).expect("a rise");
assert!(first_rise >= 2, "the wait jumped to the cap at once: {walk:?}");
// ── The case a blind cap gets WRONG ──────────────────────────────────────
//
// `AlwaysFull` sources wait a DAY between reads, because every read transfers
// everything. Capping the backoff at one hour without the `max` would make a
// FAILING full-catalogue crawl hit the source 24× harder than a healthy one.
// Backoff that speeds up on failure is worse than no backoff at all.
const DAILY: u64 = 86_400;
for n in 0..20 {
assert!(
backoff_secs(DAILY, n) >= DAILY,
"a failing daily crawl must never poll faster than a healthy one \
(failures={n} gave {})",
backoff_secs(DAILY, n),
);
}
// And an absurd failure count is answered, not crashed on. A plain `<<` here
// PANICS ("attempt to shift left with overflow", measured) once the count passes
// 64, which would take down the daemon thread over arithmetic about how long to
// sleep. `checked_shl` is what makes this line answer instead.
assert_eq!(backoff_secs(QUERY_WINDOW, u32::MAX), BACKOFF_CAP_SECS);
}
/// **A daemon's tick follows the source's DECLARED freshness, not one flat hour.**
///
/// `refresh_secs` defaulted to 3600 for every daemon in the roster. That single
/// number is wrong in both directions at once, and it is wrong by a factor of
/// twelve and twenty-four respectively: `polisen-events` is a query window over
/// incidents that a duty officer expects to see within minutes, and an hour-old
/// incident feed is a stale one; the DCAT catalogue and the PxWeb table index are
/// AlwaysFull re-reads of a registry that changes a few times a year, and re-reading
/// them every hour is 24 pointless full fetches a day, per collector, forever.
///
/// The declaration already exists — every registry `Collector` publishes a
/// `Freshness` — and after the daemon/registry name drift was fixed (adc4bc3) a
/// daemon can finally REACH its twin's declaration by name. That is the whole
/// reason the rename had to come first.
///
/// RED before the fix: every one of these read 3600.
#[test]
fn a_daemons_tick_follows_the_freshness_its_twin_declared() {
// polisen-events: a QUERY WINDOW. Minutes, not hours.
assert_eq!(
refresh_secs(&serde_json::json!({}), "polisen-events"),
300,
"an incident feed is a query window and must be read in minutes",
);
// dcat-ap-se: an ALWAYS-FULL registry crawl. Once a day is generous.
assert_eq!(
refresh_secs(&serde_json::json!({}), "dcat-ap-se"),
86_400,
"a full re-read of the national catalogue is a daily job, not an hourly one",
);
// The two must actually DIFFER, or this test would pass on any flat constant —
// which is precisely the bug it exists to catch.
assert_ne!(
refresh_secs(&serde_json::json!({}), "polisen-events"),
refresh_secs(&serde_json::json!({}), "dcat-ap-se"),
"the cadence is not a function of the source if every source gets the same one",
);
// An operator's own `refresh_secs` still wins over the declaration: the
// freshness is a DEFAULT, not a ceiling.
assert_eq!(
refresh_secs(&serde_json::json!({ "refresh_secs": 42 }), "polisen-events"),
42,
"an explicit refresh_secs is the operator's call and outranks the default",
);
// And a name nothing answers to still gets a number rather than a panic — the
// honest one for a source nobody has described, which is `AlwaysFull`'s.
assert_eq!(
refresh_secs(&serde_json::json!({}), "no-such-collector"),
crate::freshness::Freshness::default().default_refresh_secs(),
"an undescribed source re-reads everything, so it re-reads it rarely",
);
}
use crate::daemon::{
BiosLog, EnvSecretStore, Fetcher, LogSink, MatrixColor, MemWarehouse, PhaseSink, SecretStore,
SecretString, Shutdown, WarehouseSink, matrix_color,
};
use std::collections::BTreeMap;
/// A test [`SecretStore`] that answers a fixed value for ANY name (or nothing) — the
/// injected fake that proves the trait actually dispatches (and that `None` goes RED).
struct FakeSecretStore(Option<String>);
impl SecretStore for FakeSecretStore {
fn secret(&self, _name: &str) -> Option<SecretString> {
self.0.clone().map(SecretString::new)
}
}
/// Build a trafikverket ctx with an injected `secret` store and a fetch seam that
/// CAPTURES the POST body (so a test can prove the resolved secret reached the wire),
/// returning the synthetic reply. Also returns a log recorder to prove no leak.
fn trafik_ctx(
secret: std::sync::Arc<dyn SecretStore>,
) -> (
CollectorCtx,
Arc<MemWarehouse>,
Arc<Mutex<Vec<String>>>, // captured POST bodies
Arc<Mutex<Vec<String>>>, // log lines
BiosLog,
) {
let url = "https://api.trafikinfo.trafikverket.se/v2/data.json";
let reply = crate::sample::SYNTHETIC_TRAFIKVERKET_JSON.as_bytes().to_vec();
let bodies: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
let bodies_w = bodies.clone();
let want_url = url.to_string();
let fetch = Fetcher::from_verbs(
move |u| anyhow::bail!("unexpected GET {u}"),
move |u, body, _ct| {
assert_eq!(u, want_url, "POST hits the Trafikinfo endpoint");
bodies_w
.lock()
.unwrap_or_else(|e| e.into_inner())
.push(body.to_string());
Ok(reply.clone())
},
);
let warehouse = Arc::new(MemWarehouse::new());
let (log, lines) = LogSink::recorder();
let bios = BiosLog::new();
let ctx = CollectorCtx {
name: "trafikverket-trainpos".into(),
disk_dir: std::env::temp_dir(),
config: serde_json::json!({}),
bucket: Arc::new(crate::daemon::MemObjectStore::new()),
warehouse: warehouse.clone(),
fetch,
phase: PhaseSink::noop(),
log,
bios: bios.bus(),
shutdown: Shutdown::new(),
secret,
marks: std::sync::Arc::new(crate::freshness::MemMarkStore::new()),
nudge: crate::daemon::Nudge::new(),
jobs: <dyn crate::daemon::ChildJobs>::noop(),
};
(ctx, warehouse, bodies, lines, bios)
}
/// Build a ctx whose fetch seam serves `bytes` for `url` (hermetic — no network),
/// with an in-memory warehouse + a phase recorder + a BIOS log, all returned so
/// the test can assert every seam.
fn fixture_ctx(
name: &str,
url: &str,
bytes: &[u8],
) -> (
CollectorCtx,
Arc<MemWarehouse>,
Arc<Mutex<Vec<Phase>>>,
BiosLog,
) {
let mut map = BTreeMap::new();
map.insert(url.to_string(), bytes.to_vec());
let warehouse = Arc::new(MemWarehouse::new());
let (phase_sink, phases) = PhaseSink::recorder();
let bios = BiosLog::new();
let ctx = CollectorCtx {
name: name.into(),
disk_dir: std::env::temp_dir(),
config: serde_json::json!({}),
bucket: Arc::new(crate::daemon::MemObjectStore::new()),
warehouse: warehouse.clone(),
fetch: Fetcher::from_map(map),
phase: phase_sink,
log: LogSink::noop(),
bios: bios.bus(),
shutdown: Shutdown::new(),
secret: std::sync::Arc::new(crate::daemon::EnvSecretStore::new()),
marks: std::sync::Arc::new(crate::freshness::MemMarkStore::new()),
nudge: crate::daemon::Nudge::new(),
jobs: <dyn crate::daemon::ChildJobs>::noop(),
};
(ctx, warehouse, phases, bios)
}
fn assert_reached_online(
daemon: &dyn Collector,
warehouse: &MemWarehouse,
phases: &Arc<Mutex<Vec<Phase>>>,
bios: &BiosLog,
table: &str,
name: &str,
) {
assert_eq!(daemon.phase(), Phase::Online, "{name}: reached ONLINE");
assert!(daemon.healthy(), "{name}: Online is healthy");
assert_eq!(
matrix_color(daemon),
MatrixColor::Green,
"{name}: matrix cell green"
);
assert!(
warehouse.rows(table) > 0,
"{name}: rows landed in `{table}` (empty-is-RED otherwise)"
);
let seq = phases.lock().unwrap().clone();
assert!(
seq.iter().any(|p| matches!(p, Phase::Downloading { .. })),
"{name}: went through Downloading: {seq:?}"
);
assert!(
seq.iter().any(|p| matches!(p, Phase::Indexing)),
"{name}: went through Indexing: {seq:?}"
);
assert_eq!(seq.last(), Some(&Phase::Online), "{name}: ends Online");
assert!(
bios.is_online(name),
"{name}: tuned in to ONLINE on the BIOS bus"
);
}
/// **dcat reaches ONLINE and lands registry rows in Iceberg.** The offline-safe
/// synthetic DCAT-AP-SE RDF/XML (2 datasets) is served through the fetch seam; the
/// daemon fetches → parses → appends to `dcat_registry` → Online.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn dcat_daemon_reaches_online_and_sinks_registry() {
let daemon = DcatDaemon::with_config(true, DcatDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"dcat",
&url,
crate::sample::SYNTHETIC_DCAT_RDFXML.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(&daemon, &wh, &phases, &bios, "dcat_registry", "dcat");
// The facet the matrix expand renders carries the provenance.
let facet = CollectorState::facet(&daemon);
assert_eq!(facet["sink"], "iceberg");
assert_eq!(facet["table"], "dcat_registry");
assert!(
facet["rows_written"].as_u64().unwrap() >= 2,
"≥2 registry rows: {facet}"
);
}
/// **pxweb reaches ONLINE and lands cube cells in Iceberg.** The synthetic
/// JSON-stat2 cube (2×2×1 = 4 cells) drives fetch → parse → append → Online.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn pxweb_daemon_reaches_online_and_sinks_cube() {
let daemon = PxWebDaemon::with_config(true, PxWebDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"pxweb",
&url,
crate::sample::SYNTHETIC_PXWEB_JSONSTAT2.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(&daemon, &wh, &phases, &bios, "pxweb_cube", "pxweb");
assert_eq!(wh.rows("pxweb_cube"), 4, "4 cube cells landed");
}
/// **osm-admin reaches ONLINE and lands admin units in Iceberg.** The synthetic
/// OSM admin feature array (4 boundaries + 1 dropped road) drives the run-loop; the
/// non-boundary road is filtered, so 4 admin units land in `admin_units`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn osm_admin_daemon_reaches_online_and_sinks_units() {
// osm-admin has no default URL (Overpass wired per-deployment); point it at a
// fixture URL that serves the synthetic feature array.
let cfg = serde_json::json!({ "url": "fixture://admin", "table": "admin_units" });
let daemon = OsmAdminDaemon::with_config(true, cfg);
let (ctx, wh, phases, bios) = fixture_ctx(
"osm-admin-boundaries",
"fixture://admin",
crate::sample::SYNTHETIC_OSM_ADMIN_JSON.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(&daemon, &wh, &phases, &bios, "admin_units", "osm-admin-boundaries");
assert_eq!(
wh.rows("admin_units"),
4,
"4 boundaries kept, the highway dropped"
);
}
/// **nordpool reaches ONLINE and lands spot-price rows in Iceberg.** The synthetic
/// Nord Pool day-ahead document (2 periods × 3 zones) drives fetch → parse → append
/// → Online; the period×zone matrix flattens to 6 per-(area, period) price rows.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn nordpool_daemon_reaches_online_and_sinks_prices() {
let daemon = NordPoolDaemon::with_config(true, NordPoolDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"nordpool",
&url,
crate::sample::SYNTHETIC_NORDPOOL_JSON.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(&daemon, &wh, &phases, &bios, "nordpool_prices", "nordpool");
assert_eq!(
wh.rows("nordpool_prices"),
6,
"2 periods × 3 zones = 6 spot-price rows"
);
let facet = CollectorState::facet(&daemon);
assert_eq!(facet["sink"], "iceberg");
assert_eq!(facet["table"], "nordpool_prices");
}
/// **eurostat-sdmx reaches ONLINE and lands observation rows in Iceberg.** The
/// synthetic Eurostat SDMX-JSON 2.1 message (1 series dim × 1 obs dim = 4 obs)
/// drives fetch → parse → append → Online; the series/observation split flattens to
/// 4 (geo, TIME_PERIOD, value) rows.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sdmx_daemon_reaches_online_and_sinks_observations() {
let daemon = SdmxDaemon::with_config(true, SdmxDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"eurostat-sdmx",
&url,
crate::sample::SYNTHETIC_EUROSTAT_SDMX_JSON.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(
&daemon,
&wh,
&phases,
&bios,
"eurostat_sdmx",
"eurostat-sdmx",
);
assert_eq!(
wh.rows("eurostat_sdmx"),
4,
"2 geo × 2 time = 4 observation rows"
);
}
/// **usgs-earthquakes reaches ONLINE and lands quake rows in Iceberg.** The
/// synthetic USGS GeoJSON FeatureCollection (2 Point features) drives fetch → parse
/// → append → Online; each locatable Point flattens to a quake row.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn usgs_daemon_reaches_online_and_sinks_quakes() {
let daemon = UsgsDaemon::with_config(true, UsgsDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"usgs-earthquakes",
&url,
crate::sample::SYNTHETIC_USGS_GEOJSON.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(
&daemon,
&wh,
&phases,
&bios,
"usgs_quakes",
"usgs-earthquakes",
);
assert_eq!(
wh.rows("usgs_quakes"),
2,
"two locatable Point features → two quake rows"
);
}
/// **openaq-air-quality reaches ONLINE and lands measurement rows in Iceberg.**
/// The synthetic OpenAQ `results` envelope (2 located readings) drives fetch →
/// parse → append → Online; each located reading flattens to a measurement row.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn openaq_daemon_reaches_online_and_sinks_measurements() {
let daemon = OpenAqDaemon::with_config(true, OpenAqDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"openaq-air-quality",
&url,
crate::sample::SYNTHETIC_OPENAQ_JSON.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(
&daemon,
&wh,
&phases,
&bios,
"openaq_measurements",
"openaq-air-quality",
);
assert_eq!(
wh.rows("openaq_measurements"),
2,
"two located readings → two measurement rows"
);
let facet = CollectorState::facet(&daemon);
assert_eq!(facet["sink"], "iceberg");
assert_eq!(facet["table"], "openaq_measurements");
}
/// **erddap-emodnet-physics reaches ONLINE and lands table rows in Iceberg.** The
/// synthetic ERDDAP `.json` `table` (2 NDBC-buoy rows) drives fetch → parse →
/// append → Online; the columnTypes-typed rows unpack column-wise to Arrow.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn erddap_daemon_reaches_online_and_sinks_table() {
let daemon = ErddapDaemon::with_config(true, ErddapDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"erddap-emodnet-physics",
&url,
crate::sample::SYNTHETIC_ERDDAP_JSON.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(
&daemon,
&wh,
&phases,
&bios,
"erddap_emodnet_physics",
"erddap-emodnet-physics",
);
assert_eq!(
wh.rows("erddap_emodnet_physics"),
2,
"two ERDDAP table rows landed"
);
}
/// **wfs reaches ONLINE and lands feature rows in Iceberg.** The synthetic WFS 2.0
/// GML `FeatureCollection` (a Point + a LineString member) drives fetch → parse →
/// append → Online; each member flattens to a feature row (WKT geometry + props).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn wfs_daemon_reaches_online_and_sinks_features() {
let daemon = WfsDaemon::with_config(true, WfsDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) =
fixture_ctx("wfs", &url, crate::sample::SYNTHETIC_WFS_GML.as_bytes());
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(&daemon, &wh, &phases, &bios, "wfs_features", "wfs");
assert_eq!(
wh.rows("wfs_features"),
2,
"two GML members → two feature rows"
);
}
/// **energy-charts-power reaches ONLINE and lands power rows in Iceberg.** The
/// synthetic Energy-Charts document (2 instants × 3 production types) drives fetch →
/// parse → append → Online; the parallel arrays zip/transpose to 6 (time, type,
/// power) rows.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn energy_charts_daemon_reaches_online_and_sinks_power() {
let daemon = EnergyChartsDaemon::with_config(true, EnergyChartsDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"energy-charts-power",
&url,
crate::sample::SYNTHETIC_ENERGY_CHARTS_JSON.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(
&daemon,
&wh,
&phases,
&bios,
"energy_charts_power",
"energy-charts-power",
);
assert_eq!(
wh.rows("energy_charts_power"),
6,
"2 instants × 3 types = 6 transposed power rows"
);
let facet = CollectorState::facet(&daemon);
assert_eq!(facet["sink"], "iceberg");
assert_eq!(facet["table"], "energy_charts_power");
}
/// **rss-atom reaches ONLINE and lands feed rows in Iceberg.** The synthetic RSS 2.0
/// feed (2 `<item>`s) drives fetch → parse → append → Online; each item flattens to
/// a feed row.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn rss_daemon_reaches_online_and_sinks_feed() {
let daemon = RssDaemon::with_config(true, RssDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"rss-atom",
&url,
crate::sample::SYNTHETIC_RSS_XML.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(&daemon, &wh, &phases, &bios, "feed_items", "rss-atom");
assert_eq!(wh.rows("feed_items"), 2, "two RSS items → two feed rows");
}
/// **entsoe reaches ONLINE and lands market-document rows in Iceberg.** The synthetic
/// ENTSO-E `Publication_MarketDocument` (SE3, 2 quantity Points) drives fetch → parse
/// → append → Online; each TimeSeries/Period/Point flattens to a row.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn entsoe_daemon_reaches_online_and_sinks_points() {
let daemon = EntsoeDaemon::with_config(true, EntsoeDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"entsoe",
&url,
crate::sample::SYNTHETIC_ENTSOE_XML.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(&daemon, &wh, &phases, &bios, "entsoe_prices", "entsoe");
assert_eq!(
wh.rows("entsoe_prices"),
2,
"two SE3 quantity Points → two rows"
);
}
/// **entsoe_day_ahead reaches ONLINE and lands spot-price rows in Iceberg.** The
/// synthetic A44 `Publication_MarketDocument` (SE3, 3 hourly `price.amount` Points)
/// drives fetch → parse → append → Online; each Point flattens to a price row.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn entsoe_day_ahead_daemon_reaches_online_and_sinks_prices() {
let daemon =
EntsoeDayAheadDaemon::with_config(true, EntsoeDayAheadDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"entsoe_day_ahead",
&url,
crate::sample::SYNTHETIC_ENTSOE_A44_XML.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(
&daemon,
&wh,
&phases,
&bios,
"entsoe_day_ahead_prices",
"entsoe_day_ahead",
);
assert_eq!(
wh.rows("entsoe_day_ahead_prices"),
3,
"three A44 price.amount Points → three price rows"
);
}
/// **entsoe_generation reaches ONLINE and lands generation rows in Iceberg.** The
/// synthetic A75 `GL_MarketDocument` (SE3; psrType B16=Solar + B19=Wind Onshore, 2
/// quantity Points each) drives fetch → parse → append → Online; each Point flattens
/// to a row carrying the production type.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn entsoe_generation_daemon_reaches_online_and_sinks_generation() {
let daemon =
EntsoeGenerationDaemon::with_config(true, EntsoeGenerationDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"entsoe_generation",
&url,
crate::sample::SYNTHETIC_ENTSOE_A75_XML.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(
&daemon,
&wh,
&phases,
&bios,
"entsoe_generation",
"entsoe_generation",
);
assert_eq!(
wh.rows("entsoe_generation"),
4,
"2 series × 2 quantity Points = 4 generation rows"
);
}
/// **gtfs-rt-vehicles reaches ONLINE and lands vehicle rows in Iceberg.** The
/// synthetic GTFS-Realtime protobuf `FeedMessage` (a `FeedHeader` + two locatable
/// vehicles) is served through the fetch seam; the daemon fetches → decodes the
/// protobuf → appends to `gtfs_rt_vehicles` → Online. This is the crate's one
/// continuous stream, proven headless with no live agency feed.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn gtfs_rt_daemon_reaches_online_and_sinks_vehicles() {
// gtfs-rt has no default URL (per-agency `.pb`); point it at a fixture URL
// that serves the synthetic protobuf feed.
let cfg = serde_json::json!({ "url": "fixture://gtfs-rt", "table": "gtfs_rt_vehicles" });
let daemon = GtfsRtVehiclesDaemon::with_config(true, cfg);
let pb = crate::sample::synthetic_gtfs_rt_pb();
let (ctx, wh, phases, bios) = fixture_ctx("gtfs-rt-vehicles", "fixture://gtfs-rt", &pb);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(
&daemon,
&wh,
&phases,
&bios,
"gtfs_rt_vehicles",
"gtfs-rt-vehicles",
);
assert_eq!(
wh.rows("gtfs_rt_vehicles"),
2,
"two locatable vehicles → two position rows"
);
let facet = CollectorState::facet(&daemon);
assert_eq!(facet["sink"], "iceberg");
assert_eq!(facet["table"], "gtfs_rt_vehicles");
}
/// **gtfs-stops reaches ONLINE and lands stop rows in Iceberg.** A synthetic GTFS
/// static feed zip (the offline-safe `stops.txt` + `routes.txt` members, DEFLATE)
/// is served through the fetch seam; the daemon fetches → unzips + parses the
/// `stops.txt` member → appends to `gtfs_stops` → Online. Gated behind `gtfs`
/// (the pure-Rust ZIP + csv reader) exactly as the daemon is.
#[cfg(feature = "gtfs")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn gtfs_stops_daemon_reaches_online_and_sinks_stops() {
// A DEFLATE (method 8) feed zip carrying the two stops + two routes fixtures;
// the daemon picks the `stops.txt` member.
let zip = crate::gtfs::build_zip(&[
(
"stops.txt",
8,
crate::sample::SYNTHETIC_GTFS_STOPS_CSV.as_bytes(),
),
(
"routes.txt",
8,
crate::sample::SYNTHETIC_GTFS_ROUTES_CSV.as_bytes(),
),
]);
let cfg = serde_json::json!({ "url": "fixture://gtfs", "table": "gtfs_stops" });
let daemon = GtfsStopsDaemon::with_config(true, cfg);
let (ctx, wh, phases, bios) = fixture_ctx("gtfs-stops", "fixture://gtfs", &zip);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(&daemon, &wh, &phases, &bios, "gtfs_stops", "gtfs-stops");
assert_eq!(
wh.rows("gtfs_stops"),
2,
"two GTFS stops → two stop-point rows"
);
let facet = CollectorState::facet(&daemon);
assert_eq!(facet["sink"], "iceberg");
assert_eq!(facet["table"], "gtfs_stops");
}
/// **worldbank reaches ONLINE and lands statistics rows in Iceberg.** The synthetic
/// World Bank `[meta, data]` envelope (2 population observations) drives fetch →
/// parse → append → Online; the `/1` observation array flattens to 2 statistics rows.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn worldbank_daemon_reaches_online_and_sinks_indicators() {
let daemon = WorldBankDaemon::with_config(true, WorldBankDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"worldbank",
&url,
crate::sample::SYNTHETIC_WORLDBANK_JSON.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(
&daemon,
&wh,
&phases,
&bios,
"worldbank_indicators",
"worldbank",
);
assert_eq!(
wh.rows("worldbank_indicators"),
2,
"two World Bank observations → two rows"
);
let facet = CollectorState::facet(&daemon);
assert_eq!(facet["sink"], "iceberg");
assert_eq!(facet["table"], "worldbank_indicators");
}
/// **gbif reaches ONLINE and lands occurrence rows in Iceberg.** The synthetic GBIF
/// `results[]` envelope (2 occurrences) drives fetch → parse → append → Online; each
/// occurrence flattens to a biodiversity point row.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn gbif_daemon_reaches_online_and_sinks_occurrences() {
let daemon = GbifDaemon::with_config(true, GbifDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) =
fixture_ctx("gbif", &url, crate::sample::SYNTHETIC_GBIF_JSON.as_bytes());
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(&daemon, &wh, &phases, &bios, "gbif_occurrences", "gbif");
assert_eq!(
wh.rows("gbif_occurrences"),
2,
"two GBIF occurrences → two rows"
);
}
/// **eea-wise-water reaches ONLINE and lands water-quality rows in Iceberg.** The
/// synthetic EEA WISE Discodata `results[]` envelope (2 aggregated readings) drives
/// fetch → parse → append → Online; each reading flattens to a water-quality row.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn eea_wise_daemon_reaches_online_and_sinks_water_quality() {
let daemon = EeaWiseDaemon::with_config(true, EeaWiseDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"eea-wise-water",
&url,
crate::sample::SYNTHETIC_EEA_WISE_JSON.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(&daemon, &wh, &phases, &bios, "eea_wise", "eea-wise-water");
assert_eq!(
wh.rows("eea_wise"),
2,
"two EEA WISE readings → two water-quality rows"
);
}
// ── WAVE 2, batch 6: polisen / smhi-weather / trafikverket-trainpos ─────────────
/// **polisen-events reaches ONLINE and lands event rows.** The offline-safe synthetic
/// Polisen events array (2 incidents, records at the document root) drives fetch →
/// parse → append → Online; both incidents land in `polisen_events`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn polisen_daemon_reaches_online_and_sinks_events() {
let daemon = PolisenDaemon::with_config(true, PolisenDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"polisen-events",
&url,
crate::sample::SYNTHETIC_POLISEN_JSON.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(&daemon, &wh, &phases, &bios, "polisen_events", "polisen-events");
assert_eq!(wh.rows("polisen_events"), 2, "two Polisen incidents landed");
let facet = CollectorState::facet(&daemon);
assert_eq!(facet["sink"], "iceberg");
assert_eq!(facet["table"], "polisen_events");
}
/// **O2c, the gate: tick 2 lands NOTHING and says why.**
///
/// This is what the whole seam was for. `polisen-events` declares a `QueryWindow`,
/// so a second tick against an unchanged source must land 0 rows with an
/// *explained* reason — not re-land the same 2 incidents, which is what the
/// blind-append loop did every hour for every daemon in the roster.
///
/// It is red in three separate ways if the wiring is wrong, and each one matters:
/// tick 1 landing something other than 2 means the tick went LIVE (500 is the
/// polisen page size, and that is exactly the failure that blocked this row on
/// 2026-09-04); tick 2 landing 2 more means the mark never advanced; tick 2 going
/// `Failed` means an explained empty was mistaken for empty-is-RED.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_second_polisen_tick_lands_nothing_and_says_why() {
let daemon = PolisenDaemon::with_config(true, PolisenDaemon::default_config());
let url = daemon.url();
let (ctx, wh, _phases, _bios) = fixture_ctx(
"polisen-events",
&url,
crate::sample::SYNTHETIC_POLISEN_JSON.as_bytes(),
);
daemon.reconcile_once(&ctx).await.expect("tick 1");
assert_eq!(
wh.rows("polisen_events"),
2,
"tick 1 lands the TWO canned incidents — 500 would mean the tick reached \
polisen.se and the offline seam is not wired",
);
assert_eq!(daemon.phase(), Phase::Online, "tick 1 is Online");
assert!(
!ctx.marks.all().is_empty(),
"tick 1 advanced a read mark — without one, tick 2 has nothing to be \
incremental ABOUT",
);
daemon.reconcile_once(&ctx).await.expect("tick 2");
assert_eq!(
wh.rows("polisen_events"),
2,
"tick 2 lands NOTHING — the same two incidents are not re-landed",
);
assert_eq!(
daemon.phase(),
Phase::Online,
"an EXPLAINED empty is Online, not Failed: the source proved unchanged, \
which is a success, and empty-is-RED must not swallow it. Got {:?}",
daemon.phase(),
);
assert!(
daemon.healthy(),
"a daemon that correctly skipped an unchanged source is healthy",
);
}
/// **smhi-weather reaches ONLINE and lands station rows.** The synthetic SMHI
/// parameter station-set (2 stations at `/station`) drives fetch → parse → append →
/// Online; both stations land in `smhi_weather`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn smhi_weather_daemon_reaches_online_and_sinks_stations() {
let daemon = SmhiWeatherDaemon::with_config(true, SmhiWeatherDaemon::default_config());
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"smhi-weather",
&url,
crate::sample::SYNTHETIC_SMHI_WEATHER_JSON.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(&daemon, &wh, &phases, &bios, "smhi_weather", "smhi-weather");
assert_eq!(wh.rows("smhi_weather"), 2, "two SMHI stations landed");
}
/// **trafikverket-trainpos reaches ONLINE via a POST-body query and lands position
/// rows.** The daemon builds an XML `<REQUEST>` (key from config) and POSTs it; the
/// offline `from_map` fixture serves the synthetic `data.json` reply by URL (the body
/// is ignored), proving the POST path headless. The one TrainPosition record lands in
/// `trafikverket_trainpos` — a moving-map position row (wgs84 + timestamp + speed).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn trafikverket_daemon_reaches_online_via_post_and_sinks_positions() {
// A dummy key IN CONFIG — no env var, no network. The fixture ignores the body.
let cfg = serde_json::json!({
"url": "https://api.trafikinfo.trafikverket.se/v2/data.json",
"table": "trafikverket_trainpos",
"objecttype": "TrainPosition",
"key": "offline-test-key",
});
let daemon = TrafikverketDaemon::with_config(true, cfg);
let url = daemon.url();
let (ctx, wh, phases, bios) = fixture_ctx(
"trafikverket-trainpos",
&url,
crate::sample::SYNTHETIC_TRAFIKVERKET_JSON.as_bytes(),
);
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile to Online");
assert_reached_online(
&daemon,
&wh,
&phases,
&bios,
"trafikverket_trainpos",
"trafikverket-trainpos",
);
assert_eq!(wh.rows("trafikverket_trainpos"), 1, "one train position landed");
// Prove the moving-map column survived the POST → parse → sink round trip.
let batches = wh.batches("trafikverket_trainpos");
let batch = batches.first().expect("a landed batch");
let wgs = batch
.column_by_name("wgs84")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("wgs84 utf8 column");
assert_eq!(wgs.value(0), "POINT (16.18 59.61)");
}
/// **empty-is-RED for the batch-6 daemons** — each fed a well-formed but empty source
/// (an empty array / empty result) parses to ZERO rows and paints a loud
/// [`Phase::Failed`], never a silent green. Covers the GET pair AND the POST daemon.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn batch6_empty_sources_are_red_not_silently_green() {
// polisen: an empty top-level array.
{
let daemon = PolisenDaemon::with_config(true, PolisenDaemon::default_config());
let url = daemon.url();
let (ctx, wh, _p, bios) = fixture_ctx("polisen-events", &url, b"[]");
daemon.reconcile_once(&ctx).await.expect("fact, not panic");
assert!(
matches!(daemon.phase(), Phase::Failed { .. }),
"polisen 0 rows ⇒ Failed: {:?}",
daemon.phase()
);
assert_eq!(matrix_color(&daemon), MatrixColor::Red);
assert!(!daemon.healthy());
assert_eq!(wh.rows("polisen_events"), 0, "nothing landed");
assert!(!bios.is_online("polisen-events"));
let facet = CollectorState::facet(&daemon);
assert!(
facet["error"].as_str().unwrap().contains("empty-is-RED"),
"loud reason: {facet}"
);
}
// smhi-weather: a parameter set with an empty station array.
{
let daemon = SmhiWeatherDaemon::with_config(true, SmhiWeatherDaemon::default_config());
let url = daemon.url();
let (ctx, wh, _p, bios) =
fixture_ctx("smhi-weather", &url, br#"{"key":"1","station":[]}"#);
daemon.reconcile_once(&ctx).await.expect("fact, not panic");
assert!(matches!(daemon.phase(), Phase::Failed { .. }));
assert_eq!(matrix_color(&daemon), MatrixColor::Red);
assert_eq!(wh.rows("smhi_weather"), 0);
assert!(!bios.is_online("smhi-weather"));
}
// trafikverket: a valid envelope with an empty TrainPosition result array.
{
let cfg = serde_json::json!({ "key": "offline-test-key" });
let daemon = TrafikverketDaemon::with_config(true, cfg);
let url = daemon.url();
let (ctx, wh, _p, bios) = fixture_ctx(
"trafikverket-trainpos",
&url,
br#"{"RESPONSE":{"RESULT":[{"TrainPosition":[]}]}}"#,
);
daemon.reconcile_once(&ctx).await.expect("fact, not panic");
assert!(matches!(daemon.phase(), Phase::Failed { .. }));
assert_eq!(matrix_color(&daemon), MatrixColor::Red);
assert_eq!(wh.rows("trafikverket_trainpos"), 0);
assert!(!bios.is_online("trafikverket-trainpos"));
}
}
// ── The SecretStore seam (red-first) ───────────────────────────────────────────
/// **A missing secret is RED, never a silent green.** No config `key`, and the injected
/// SecretStore answers `None` — the daemon must NOT fetch and must paint a loud `Failed`
/// naming the secret. (No `key` in config, so resolution goes purely through the store.)
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn missing_secret_is_red_not_silently_green() {
let daemon = TrafikverketDaemon::with_config(true, serde_json::json!({})); // no key
let (ctx, wh, bodies, lines, bios) =
trafik_ctx(std::sync::Arc::new(FakeSecretStore(None)));
daemon.reconcile_once(&ctx).await.expect("fact, not panic");
assert!(
matches!(daemon.phase(), Phase::Failed { .. }),
"missing secret ⇒ Failed: {:?}",
daemon.phase()
);
assert_eq!(matrix_color(&daemon), MatrixColor::Red);
assert!(!daemon.healthy());
assert!(
bodies.lock().unwrap().is_empty(),
"the daemon never POSTed without its key"
);
assert_eq!(wh.rows("trafikverket_trainpos"), 0, "nothing landed");
assert!(!bios.is_online("trafikverket-trainpos"));
let facet = CollectorState::facet(&daemon);
let err = facet["error"].as_str().unwrap();
assert!(
err.contains("TRAFIKVERKET_API_KEY") && err.contains("RED"),
"loud, secret-naming reason: {err}"
);
// The reason names the SECRET NAME, never a value (there is none) — and nothing leaked.
assert!(
lines.lock().unwrap().iter().all(|l| !l.contains("SECRET")),
"no log line carries a secret value"
);
}
/// **An empty fake also fails** — a store that answers `Some("")` is no key at all.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn empty_secret_from_store_is_red() {
let daemon = TrafikverketDaemon::with_config(true, serde_json::json!({}));
let (ctx, wh, bodies, _lines, _bios) =
trafik_ctx(std::sync::Arc::new(FakeSecretStore(Some(String::new()))));
daemon.reconcile_once(&ctx).await.expect("fact, not panic");
assert!(matches!(daemon.phase(), Phase::Failed { .. }), "empty key ⇒ Failed");
assert_eq!(matrix_color(&daemon), MatrixColor::Red);
assert!(bodies.lock().unwrap().is_empty(), "no POST with an empty key");
assert_eq!(wh.rows("trafikverket_trainpos"), 0);
}
/// **The trait dispatches: an injected fake secret reaches the POST body.** No config
/// `key`; the SecretStore hands back `SECRET-XYZ`; the capturing fetcher records the
/// XML `<REQUEST>` that went to the wire — and it carries the injected key in the
/// `<LOGIN authenticationkey=…>`. This proves resolution routes THROUGH the store (a
/// VaultSecretStore would land here identically), reaches Online and lands the row.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn injected_secret_reaches_the_request_and_reaches_online() {
let daemon = TrafikverketDaemon::with_config(true, serde_json::json!({})); // no config key
let (ctx, wh, bodies, lines, _bios) = trafik_ctx(std::sync::Arc::new(FakeSecretStore(
Some("SECRET-XYZ".to_string()),
)));
daemon.reconcile_once(&ctx).await.expect("reconcile to Online");
assert_eq!(daemon.phase(), Phase::Online, "reached Online with the injected key");
assert!(daemon.healthy());
assert_eq!(wh.rows("trafikverket_trainpos"), 1, "one train position landed");
let bodies = bodies.lock().unwrap();
assert_eq!(bodies.len(), 1, "exactly one POST went out");
assert!(
bodies[0].contains("authenticationkey=\"SECRET-XYZ\""),
"the injected secret reached the request body: {}",
bodies[0]
);
// The secret is on the WIRE (correct) but must NOT be in any LOG line.
assert!(
lines.lock().unwrap().iter().all(|l| !l.contains("SECRET-XYZ")),
"no log line leaked the secret: {:?}",
lines.lock().unwrap()
);
}
/// **The `SecretString` wrapper never renders its value.** `Debug`/`Display` are a fixed
/// redaction, and only `expose()` reveals the bytes — so a stray `{:?}`/`{}` on a config
/// or a panic message cannot leak the key.
#[test]
fn secret_string_debug_and_display_are_redacted() {
let s = SecretString::new("topsecret-value");
assert_eq!(format!("{s:?}"), "SecretString(***redacted***)");
assert_eq!(format!("{s}"), "***redacted***");
assert!(!format!("{s:?}").contains("topsecret"));
assert!(!format!("{s}").contains("topsecret"));
assert_eq!(s.expose(), "topsecret-value", "expose() is the only reveal");
}
/// **The default `EnvSecretStore` resolves env / inline, `None` otherwise** — today's
/// behavior, unchanged, and the floor a vault store replaces.
#[test]
fn env_secret_store_resolves_inline_then_env_else_none() {
let store = EnvSecretStore::new();
assert!(
store.secret("KORP_COLLECTORS_DEFINITELY_UNSET_XYZ").is_none(),
"an unset env var ⇒ None (the RED signal)"
);
let mut inline = BTreeMap::new();
inline.insert("MY_KEY".to_string(), "from-inline".to_string());
inline.insert("BLANK".to_string(), String::new());
let store = EnvSecretStore::with_inline(inline);
assert_eq!(
store.secret("MY_KEY").map(|s| s.expose().to_string()),
Some("from-inline".to_string()),
"inline literal resolves"
);
assert!(
store.secret("BLANK").is_none(),
"a blank inline literal never shadows a real value ⇒ None"
);
}
/// **empty-is-RED** (design law): an enabled collector that fetches+parses to ZERO
/// rows is a loud [`Phase::Failed`] (red), never a silent green — and the BIOS bus
/// never reports it ONLINE. Here dcat is fed an RDF doc with no `dcat:Dataset`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn empty_source_is_red_not_silently_green() {
const EMPTY_RDF: &str = r#"<?xml version="1.0"?><rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"></rdf:RDF>"#;
let daemon = DcatDaemon::with_config(true, DcatDaemon::default_config());
let url = daemon.url();
let (ctx, wh, _phases, bios) = fixture_ctx("dcat", &url, EMPTY_RDF.as_bytes());
daemon
.reconcile_once(&ctx)
.await
.expect("reconcile returns (a fact, not a panic)");
assert!(
matches!(daemon.phase(), Phase::Failed { .. }),
"0 rows ⇒ Failed: {:?}",
daemon.phase()
);
assert_eq!(
matrix_color(&daemon),
MatrixColor::Red,
"empty enabled collector paints RED"
);
assert!(!daemon.healthy());
assert_eq!(wh.rows("dcat_registry"), 0, "nothing landed");
assert!(!bios.is_online("dcat"), "never tuned in to ONLINE");
let facet = CollectorState::facet(&daemon);
assert!(
facet["error"].as_str().unwrap().contains("empty-is-RED"),
"loud reason: {facet}"
);
}
/// **A disabled collector is grey (off), never red** — even with a would-be-empty
/// source. The matrix colour keys off `enabled`, so an off collector is never a
/// failure. (Mirrors the daemon.rs matrix-colour law for these new daemons.)
#[test]
fn disabled_daemon_is_off_not_red() {
let daemon = DcatDaemon::new(); // enabled = false
assert!(!daemon.enabled());
assert_eq!(
matrix_color(&daemon),
MatrixColor::Off,
"disabled ⇒ grey/off"
);
}
/// **The config schema each Settings pane renders** carries the editable fields
/// (source URL + table), so a generic Settings pane renders these collectors
/// without knowing them — the same generic face `osm-cut` uses.
#[test]
fn schemas_expose_url_and_table_fields() {
for d in [
Arc::new(DcatDaemon::new()) as Arc<dyn Collector>,
Arc::new(PxWebDaemon::new()),
Arc::new(OsmAdminDaemon::new()),
Arc::new(PolisenDaemon::new()),
Arc::new(SmhiWeatherDaemon::new()),
Arc::new(TrafikverketDaemon::new()),
] {
let schema = d.schema();
assert!(
schema.fields.iter().any(|f| f.key == "url"),
"{}: has a url field",
d.name()
);
assert!(
schema.fields.iter().any(|f| f.key == "table"),
"{}: has a table field",
d.name()
);
}
}
/// **The Config face carries a title/description/defaults** so korp renders the
/// Settings section + matrix cell registry-driven (label travels with the collector,
/// not a korp-side name map). Each WAVE-1 daemon has a non-empty title + blurb and its
/// `defaults()` seeds the canonical source URL / table.
#[test]
fn config_face_exposes_title_description_defaults() {
for d in [
Arc::new(DcatDaemon::new()) as Arc<dyn Collector>,
Arc::new(PxWebDaemon::new()),
Arc::new(OsmAdminDaemon::new()),
Arc::new(PolisenDaemon::new()),
Arc::new(SmhiWeatherDaemon::new()),
Arc::new(TrafikverketDaemon::new()),
] {
assert!(
d.title().contains(d.name()),
"{}: title carries the name: {}",
d.name(),
d.title()
);
assert!(
!d.description().is_empty(),
"{}: has a Settings blurb",
d.name()
);
let defaults = d.defaults();
assert!(
defaults.get("table").and_then(|v| v.as_str()).is_some(),
"{}: defaults seed a table",
d.name()
);
}
// dcat's defaults seed the canonical dataportal.se endpoint.
assert_eq!(
DcatDaemon::new().defaults()["url"],
"https://admin.dataportal.se/all.rdf"
);
}
/// **The never-ending loop is stoppable** (design §5): pre-trip shutdown, run the
/// full daemon loop, and assert it reconciles once to Online then returns promptly.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn daemon_loop_is_stoppable() {
let daemon = Arc::new(PxWebDaemon::with_config(
true,
PxWebDaemon::default_config(),
));
let url = daemon.url();
let (mut ctx, _wh, _phases, _bios) = fixture_ctx(
"pxweb",
&url,
crate::sample::SYNTHETIC_PXWEB_JSONSTAT2.as_bytes(),
);
let shutdown = Shutdown::new();
shutdown.stop(); // toggled off before start
ctx.shutdown = shutdown;
tokio::time::timeout(std::time::Duration::from_secs(30), daemon.clone().run(ctx))
.await
.expect("stoppable daemon returns");
assert_eq!(
daemon.phase(),
Phase::Online,
"reconciled once to Online before stopping"
);
}
}