facett-korp/src/lib.rs
//! **facett-korp** — the korp **Ingest** pane as a facett-Elm component.
//!
//! korp's collector surface (the open-data ingest tab) lived welded inside korp's
//! 3500-line `main.rs`, so it was not drivable by the robot harness. This crate
//! extracts the minimal *drivable* pane: a collector **roster** (one row per
//! registered collector — `name`, `source → target`, `direction`) with a
//! selection and a **Run** action.
//!
//! ## The Elm seam (and why there is no `korp` / `korp-collectors` dependency)
//!
//! korp *depends on this crate* to render its Ingest view, so this crate must NOT
//! depend on `korp` (cycle), and it stays a pure leaf by NOT depending on
//! `korp-collectors` either — it mirrors the collector-descriptor shape as its own
//! serializable [`IngestRow`]. The korp host maps
//! `korp_collectors::registry_descriptors()` → `Vec<IngestRow>` (one call to
//! [`IngestRow::from_descriptor`], which preserves the collector's declared flow
//! direction) and feeds it in via [`IngestPane::from_descriptors`]. Without a live
//! registry, [`IngestPane::demo_full`] mirrors `korp_collectors::registry()` so the
//! real breadth — boats (AIS), planes (OpenSky), energy (ENTSO-E), roads (OSM /
//! Trafikverket) — still renders and drives. Running a collector is a host concern:
//! [`IngestPane`] emits
//! an [`IngestEffect::RunCollector`] the host turns into a **jera** job (never run
//! here — no async, no network).
//!
//! It implements [`facett_core::Elm`]; [`facett_core::impl_facet_via_elm!`] bridges
//! it to [`facett_core::Facet`] (and thus [`facett_core::Panel`] via the blanket
//! impl), with `update_json` routed into the Elm `update` — so the headless robot
//! harness drives it through JSON messages and asserts `state_json`. One wire,
//! every backend.
use egui::{Color32, Ui};
use facett_core::Elm;
use serde::{Deserialize, Serialize};
/// **GraphPane** — the korp Graph tab's analyst control surface (Cypher query +
/// neighbour-expand L-3 + shortest-path L-4). See the [`graph`] module.
pub mod graph;
pub use graph::{
CentralityRow, CrossFilter, GraphEffect, GraphModel, GraphMsg, GraphNodeRef, GraphPane,
DEFAULT_CYPHER,
};
/// **SearchPane** — the korp global object search control surface (Palantir pillar
/// P5, gap S-1): a query box + ranked results the analyst clicks. See the [`search`]
/// module.
pub mod search;
pub use search::{SearchEffect, SearchModel, SearchMsg, SearchPane, SearchRow};
/// **TimelinePane** — the korp Timeline tab's control surface (Palantir pillar P3,
/// gaps T-2 timeline / T-3 shared time brush): a histogram of events over time with
/// a click brush that sets the shared window. See the [`timeline`] module.
pub mod timeline;
pub use timeline::{TimelineBar, TimelineEffect, TimelineModel, TimelineMsg, TimelinePane};
/// One collector row — the facett-side mirror of a
/// `(CollectorMeta, SourceTarget)` descriptor (kept local so this crate never
/// depends on `korp-collectors`).
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct IngestRow {
/// Collector name (e.g. `opensky`, `ais`, `entsoe`, `trafikverket`).
pub name: String,
/// The open-data endpoint it pulls FROM (short label).
pub source: String,
/// Where the parsed rows land (warehouse table / Arrow / map scene).
pub target: String,
/// Data-flow direction (collectors are `source`).
pub direction: String,
}
impl IngestRow {
pub fn new(
name: impl Into<String>,
source: impl Into<String>,
target: impl Into<String>,
) -> Self {
Self {
name: name.into(),
source: source.into(),
target: target.into(),
direction: "source".into(),
}
}
/// Map a host's collector descriptor into a roster row, **preserving the
/// declared flow direction** (which [`IngestRow::new`] hardcodes to
/// `"source"`). This is the one mapping the korp host uses to feed the REAL
/// registry into the pane:
///
/// ```ignore
/// let rows = korp_collectors::registry_descriptors()
/// .into_iter()
/// .map(|(meta, st)| facett_korp::IngestRow::from_descriptor(
/// meta.name, st.source.as_str(), st.label(), meta.direction.as_str()))
/// .collect();
/// let pane = facett_korp::IngestPane::from_descriptors("Ingest", rows);
/// ```
///
/// The host stringifies the descriptor fields so this crate stays a pure leaf
/// (no `korp-collectors` dependency, no cycle).
pub fn from_descriptor(
name: impl Into<String>,
source: impl Into<String>,
target: impl Into<String>,
direction: impl Into<String>,
) -> Self {
Self {
name: name.into(),
source: source.into(),
target: target.into(),
direction: direction.into(),
}
}
}
/// The complete observable state of the Ingest pane (FC-1/FC-3).
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct IngestModel {
pub title: String,
pub rows: Vec<IngestRow>,
/// Index into `rows` of the selected collector, if any.
pub selected: Option<usize>,
/// Human note describing the last action (e.g. the queued job).
pub last_action: Option<String>,
}
/// Every input the pane accepts (FC-2) — the only legal way to mutate the model.
/// Externally-tagged serde so the robot/`update_json` seam sends `{"select": 2}`
/// for [`IngestMsg::Select`] and the bare strings `"run"` / `"clear"` for the unit
/// variants.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum IngestMsg {
/// Select the collector at this row index (ignored if out of range).
Select(usize),
/// Run the selected collector (no-op if nothing is selected).
Run,
/// Clear the last-action note.
Clear,
}
/// Side work as data (FC-8): the host runs the collector as a jera job.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum IngestEffect {
/// Ask the host to run this collector (→ a `jera` job). Carries the name.
RunCollector(String),
}
/// The korp Ingest pane.
#[derive(Clone, Debug, PartialEq)]
pub struct IngestPane {
model: IngestModel,
}
impl IngestPane {
/// Build the pane from a title + the collector roster (mapped by the host from
/// `korp_collectors::registry_descriptors()`).
pub fn new(title: impl Into<String>, rows: Vec<IngestRow>) -> Self {
Self { model: IngestModel { title: title.into(), rows, selected: None, last_action: None } }
}
/// A self-contained demo roster (the real built-in korp collectors) so tests
/// and a host without a live registry still have something to drive.
pub fn demo() -> Self {
Self::new(
"Ingest",
vec![
IngestRow::new("opensky", "OpenSky REST", "flights (map:osm3d-flight)"),
IngestRow::new("ais", "AIS REST", "vessels (warehouse:ais)"),
IngestRow::new("entsoe", "ENTSO-E API", "energy (warehouse:entsoe)"),
IngestRow::new("trafikverket", "Trafikverket API", "train_positions (map)"),
],
)
}
/// Build the pane from a host's mapped descriptor rows — the **production**
/// path (`korp_collectors::registry_descriptors()` → [`IngestRow::from_descriptor`]
/// → here). Named for the seam; the roster is the live registry, not the demo.
pub fn from_descriptors(title: impl Into<String>, rows: Vec<IngestRow>) -> Self {
Self::new(title, rows)
}
/// The **full production roster** — every built-in korp collector, mirroring
/// `korp_collectors::registry()` so a host WITHOUT a live registry (or a test)
/// still shows the real breadth: the boats / planes / energy / roads feeds the
/// task calls for, not just the 4-row [`demo`](Self::demo). The live host maps
/// the real `registry_descriptors()` through [`IngestRow::from_descriptor`]; this
/// is the same shape, hand-mirrored (no `korp-collectors` dep in this leaf).
pub fn demo_full() -> Self {
Self::from_descriptors(
"Ingest",
vec![
// open-data catalogues + statistics
IngestRow::from_descriptor("dcat", "dcat-catalog", "warehouse:dcat_registry", "source"),
IngestRow::from_descriptor("pxweb", "pxweb-api", "warehouse:scb_cube", "source"),
IngestRow::from_descriptor("kolada", "generic-rest", "warehouse:kolada_kpi", "source"),
IngestRow::from_descriptor("polisen", "generic-rest", "warehouse:police_events", "source"),
IngestRow::from_descriptor("smhi_metobs", "generic-rest", "warehouse:smhi_metobs", "source"),
IngestRow::from_descriptor("smhi_weather", "generic-rest", "warehouse:smhi_forecast", "source"),
// energy
IngestRow::from_descriptor("entsoe", "entsoe", "warehouse:entsoe_load", "source"),
IngestRow::from_descriptor("entsoe_generation", "entsoe", "warehouse:entsoe_generation", "source"),
// boats (AIS)
IngestRow::from_descriptor("ais", "ais", "map-scene:vessels (scene=osm3d-flight)", "source"),
IngestRow::from_descriptor("marinetraffic_ais", "ais", "map-scene:vessels (scene=osm3d-flight)", "source"),
// planes (OpenSky)
IngestRow::from_descriptor("opensky", "generic-rest", "map-scene:flights (scene=osm3d-flight)", "source"),
IngestRow::from_descriptor("opensky_europe_bbox", "generic-rest", "map-scene:flights (scene=osm3d-flight)", "source"),
// micromobility
IngestRow::from_descriptor("gbfs_citybikes", "generic-rest", "warehouse:gbfs_stations", "source"),
// roads (OSM Sweden + Trafikverket)
IngestRow::from_descriptor("osm_overpass", "generic-rest", "map-scene:roads (scene=osm3d)", "source"),
IngestRow::from_descriptor("osm", "osm", "map-scene:roads (scene=osm3d)", "source"),
IngestRow::from_descriptor("osm_admin", "osm-admin", "warehouse:osm_boundaries", "source"),
IngestRow::from_descriptor("trafikverket_train_positions", "trafikverket", "map-scene:train_positions (scene=osm3d)", "source"),
IngestRow::from_descriptor("trafikverket_trains", "trafikverket", "warehouse:train_announcements", "source"),
IngestRow::from_descriptor("trafikverket_road_network", "trafikverket", "map-scene:roads (scene=osm3d)", "source"),
],
)
}
pub fn model(&self) -> &IngestModel {
&self.model
}
/// Index of the first collector whose `name` equals `name`, if present.
pub fn index_of(&self, name: &str) -> Option<usize> {
self.model.rows.iter().position(|r| r.name == name)
}
/// The name of the selected collector, if any.
pub fn selected_name(&self) -> Option<&str> {
self.model.selected.and_then(|i| self.model.rows.get(i)).map(|r| r.name.as_str())
}
}
impl Elm for IngestPane {
type Model = IngestModel;
type Msg = IngestMsg;
type Effect = IngestEffect;
fn title(&self) -> &str {
&self.model.title
}
fn state(&self) -> &Self::Model {
&self.model
}
fn update(&mut self, msg: Self::Msg) -> Vec<Self::Effect> {
match msg {
IngestMsg::Select(i) => {
if i < self.model.rows.len() {
self.model.selected = Some(i);
}
Vec::new()
}
IngestMsg::Run => {
if let Some(name) = self.selected_name().map(str::to_string) {
self.model.last_action = Some(format!("queued jera job: collect {name}"));
vec![IngestEffect::RunCollector(name)]
} else {
self.model.last_action = Some("no collector selected".into());
Vec::new()
}
}
IngestMsg::Clear => {
self.model.last_action = None;
Vec::new()
}
}
}
fn view(&self, ui: &mut Ui) -> Vec<Self::Msg> {
let mut msgs = Vec::new();
ui.heading("korp · Ingest");
ui.label(format!("{} collectors", self.model.rows.len())); // plain numbers
ui.separator();
for (i, row) in self.model.rows.iter().enumerate() {
let selected = self.model.selected == Some(i);
let label = format!("{} — {} → {}", row.name, row.source, row.target);
if ui.selectable_label(selected, label).clicked() {
msgs.push(IngestMsg::Select(i));
}
}
ui.separator();
ui.horizontal(|ui| {
let run = ui.add_enabled(self.model.selected.is_some(), egui::Button::new("▶ Run collector"));
if run.clicked() {
msgs.push(IngestMsg::Run);
}
if let Some(a) = &self.model.last_action {
let resp = ui.colored_label(Color32::from_rgb(120, 200, 120), a);
// The "no collector selected" degraded action carries the stable UI
// error code facet-korp-1 (facett_core::errcode) on its a11y atom so
// tests + consumers react to the CODE, not a matched string.
if a == "no collector selected" {
let a = a.clone();
resp.widget_info(|| {
facett_core::a11y::Semantics::error(a.clone()).error_code("facet-korp-1").widget_info()
});
}
}
});
msgs
}
}
facett_core::impl_facet_via_elm!(IngestPane, custom_state_json, {
fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"kind": "korp-ingest",
"title": self.model.title,
"count": self.model.rows.len(),
"selected": self.model.selected,
"selected_name": self.selected_name(),
"last_action": self.model.last_action,
"rows": self.model.rows,
// The stable UI ERROR CODE for the "no collector selected" degraded state
// (facett_core::errcode): present so tests + consumers react to the CODE.
"error_code": if self.model.last_action.as_deref() == Some("no collector selected") {
serde_json::json!("facet-korp-1")
} else {
serde_json::Value::Null
},
})
}
/// STRUCTURAL severity (the Robot-UI gate signal): RED when a Run was attempted
/// with no collector selected (the facet-korp-1 state), the Info floor otherwise.
fn severity(&self) -> facett_core::Severity {
if self.model.last_action.as_deref() == Some("no collector selected") {
facett_core::Severity::Error
} else {
facett_core::Severity::Info
}
}
fn update_json(&mut self, msg_json: &str) {
if let Ok(msg) = serde_json::from_str::<IngestMsg>(msg_json) {
let _ = <Self as facett_core::Elm>::update(self, msg);
}
}
});
#[cfg(test)]
mod tests {
use super::*;
use facett_core::Panel;
#[test]
fn demo_roster_is_well_formed() {
let p = IngestPane::demo();
assert_eq!(p.model().rows.len(), 4);
assert!(p.model().selected.is_none());
assert!(p.selected_name().is_none());
}
#[test]
fn select_then_run_emits_run_collector_effect() {
let mut p = IngestPane::demo();
assert!(p.update(IngestMsg::Select(0)).is_empty());
assert_eq!(p.selected_name(), Some("opensky"));
let fx = p.update(IngestMsg::Run);
assert_eq!(fx, vec![IngestEffect::RunCollector("opensky".into())]);
assert_eq!(p.model().last_action.as_deref(), Some("queued jera job: collect opensky"));
}
#[test]
fn run_without_selection_is_a_clean_noop() {
let mut p = IngestPane::demo();
assert!(p.update(IngestMsg::Run).is_empty());
assert_eq!(p.model().last_action.as_deref(), Some("no collector selected"));
}
#[test]
fn out_of_range_select_is_ignored() {
let mut p = IngestPane::demo();
p.update(IngestMsg::Select(99));
assert!(p.model().selected.is_none());
}
/// The **robot/remote-control seam driven IN CODE**: select + run through the
/// same `update_json` the headless robot harness uses, then assert `state_json`.
/// No pixels, no AI in the loop — self-driving.
#[test]
fn robot_json_drive_select_and_run() {
let mut p = IngestPane::demo();
Panel::update_json(&mut p, r#"{"select":1}"#);
Panel::update_json(&mut p, "\"run\"");
let st = Panel::state_json(&p);
assert_eq!(st["kind"], "korp-ingest");
assert_eq!(st["selected_name"], "ais");
assert_eq!(st["last_action"], "queued jera job: collect ais");
assert_eq!(st["count"], 4);
// malformed / unknown messages are ignored, not a panic.
Panel::update_json(&mut p, "not json [[[");
Panel::update_json(&mut p, r#"{"nonsense":true}"#);
assert_eq!(Panel::state_json(&p)["selected_name"], "ais");
}
/// The **full production roster** carries the boats / planes / energy / roads
/// collectors korp needs — and it is drivable. Selecting the AIS (boats) feed by
/// name and running it queues the jera job. Proves `from_descriptor` +
/// `demo_full` render the real breadth (not just the 4-row demo).
#[test]
fn full_roster_carries_boats_planes_energy_roads_and_is_drivable() {
let mut p = IngestPane::demo_full();
assert!(p.model().rows.len() >= 15, "full roster is the real breadth");
// the four data families the task names are present, by collector name:
for feed in ["ais", "opensky", "entsoe", "trafikverket_road_network"] {
assert!(p.index_of(feed).is_some(), "production roster carries {feed}");
}
// direction is preserved from the descriptor (not hardcoded elsewhere).
assert!(p.model().rows.iter().all(|r| r.direction == "source"));
// drive: select the boats (AIS) feed by name, then run.
let ais = p.index_of("ais").unwrap();
Panel::update_json(&mut p, &format!("{{\"select\":{ais}}}"));
Panel::update_json(&mut p, "\"run\"");
let st = Panel::state_json(&p);
assert_eq!(st["selected_name"], "ais");
assert_eq!(st["last_action"], "queued jera job: collect ais");
}
/// `from_descriptor` preserves a non-`source` direction (e.g. a sink), unlike
/// `IngestRow::new` which hardcodes `"source"`.
#[test]
fn from_descriptor_preserves_direction() {
let r = IngestRow::from_descriptor("warehouse", "arrow", "iceberg:table", "sink");
assert_eq!(r.direction, "sink");
let n = IngestRow::new("ais", "ais", "vessels");
assert_eq!(n.direction, "source");
}
/// Emit the test-matrix cell so `nornir test --features testmatrix` SEES this
/// pane's remote-drive is covered (green ⟺ nothing untested). No-op otherwise.
#[test]
fn testmatrix_cell_ingest_remote_drive() {
let mut p = IngestPane::demo();
Panel::update_json(&mut p, r#"{"select":0}"#);
Panel::update_json(&mut p, "\"run\"");
let ok = Panel::state_json(&p)["last_action"] == "queued jera job: collect opensky";
facett_core::testmatrix::emit(
"facett-korp/ingest",
"robot_json_drive_runs_selected_collector",
ok,
"select→run via update_json emits queued-jera-job in state_json",
);
assert!(ok);
}
/// Emit a matrix cell proving the **full production roster** (the collectors
/// korp needs) is present + drivable — one visible row for the boats/planes/
/// energy/roads consumption seam. No-op without `testmatrix`.
#[test]
fn testmatrix_cell_full_roster_consumed() {
let p = IngestPane::demo_full();
let ok = ["ais", "opensky", "entsoe", "trafikverket_road_network"]
.iter()
.all(|f| p.index_of(f).is_some());
facett_core::testmatrix::emit(
"facett-korp/ingest",
"full_registry_roster_consumed",
ok,
"demo_full roster (from_descriptor) carries boats/planes/energy/roads collectors",
);
assert!(ok);
}
}