src/viz/model.rs
//! Pure data model for the viz — no egui types here so we can
//! unit-test the loader without a display.
use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use anyhow::Result;
use chrono::{DateTime, Utc};
use serde::Deserialize;
use uuid::Uuid;
use crate::bench::BenchRun;
use crate::release::pipeline::{ReleaseReport, query_release_history};
use crate::warehouse::dep_graph::{DepGraphSnapshot, query_dep_graph_snapshots};
use crate::warehouse::iceberg::IcebergWarehouse;
use crate::warehouse::{BenchFilter, Warehouse};
// The `Viz.Timeline` view-model TYPES were relocated to the `nornir-wire` edda
// leaf (`nornir_wire::view::timeline`) so a read-client can decode the payload
// without a `nornir` dependency. They are pure serde data — `Timeline`'s one
// non-scalar field is a `nornir_depgraph::DepGraphSnapshot`, itself already a leaf
// (`crate::warehouse::dep_graph::DepGraphSnapshot` re-exports it), so the model
// carried no warehouse guts and moved whole (the plan's relocate tactic). The
// PRODUCERS (`build_timeline`/`load_timeline` + the git-history/bench folds) stay
// here — they still read the warehouse — and fill these re-exported types.
pub use nornir_wire::view::timeline::{BenchHistory, BenchPoint, Lane, LaneNode, Timeline};
/// Read every recorded release for `workspace_name` and group rows
/// into per-repo lanes sorted by timestamp.
/// Parse a release tag into `(crate, version)`: `skade-v0.4.8` → `("skade",
/// "0.4.8")`, `v0.1.5` → `("", "0.1.5")`, otherwise `(tag, "")`.
fn parse_tag_version(tag: &str) -> (String, String) {
if let Some(idx) = tag.rfind("-v") {
let ver = &tag[idx + 2..];
if ver.chars().next().is_some_and(|c| c.is_ascii_digit()) {
return (tag[..idx].to_string(), ver.to_string());
}
}
if let Some(ver) = tag.strip_prefix('v') {
if ver.chars().next().is_some_and(|c| c.is_ascii_digit()) {
return (String::new(), ver.to_string());
}
}
(tag.to_string(), String::new())
}
/// Read every member git repo under `git_dir` and synthesize one [`Lane`] per
/// repo from its **tags** — the whole-git-history fallback for a workspace with
/// no recorded releases. Best-effort: unreadable repos/tags are skipped.
fn git_history_lanes(git_dir: &Path) -> Vec<Lane> {
let mut lanes = Vec::new();
let Ok(rd) = std::fs::read_dir(git_dir) else {
return lanes;
};
for entry in rd.flatten() {
let repo_path = entry.path();
if !repo_path.is_dir() {
continue;
}
let repo_name = entry.file_name().to_string_lossy().to_string();
let repo = match gix::open(&repo_path) {
Ok(r) => r,
Err(e) => {
eprintln!("git-fallback: open {repo_path:?} failed: {e}");
continue;
}
};
let platform = match repo.references() {
Ok(p) => p,
Err(e) => {
eprintln!("git-fallback: references {repo_name} failed: {e}");
continue;
}
};
let tags = match platform.tags() {
Ok(t) => t,
Err(e) => {
eprintln!("git-fallback: tags {repo_name} failed: {e}");
continue;
}
};
let mut nodes = Vec::new();
for tag in tags.flatten() {
let tag_name = tag.name().shorten().to_string();
let Ok(id) = tag.into_fully_peeled_id() else {
continue;
};
let sha = id.to_hex().to_string();
let Ok(obj) = id.object() else { continue };
let Ok(commit) = obj.try_into_commit() else {
continue;
};
let secs = commit.time().map(|t| t.seconds).unwrap_or(0);
let timestamp = DateTime::<Utc>::from_timestamp(secs, 0).unwrap_or_else(Utc::now);
let (crate_name, version) = parse_tag_version(&tag_name);
let published_versions = if version.is_empty() {
Vec::new()
} else {
vec![(crate_name, version)]
};
nodes.push(LaneNode {
release_id: Uuid::new_v4(),
timestamp,
sha,
branch: tag_name,
dirty: false,
gate_status: "succeeded".to_string(),
tests_passed: 0,
tests_failed: 0,
published_versions,
});
}
// Success-path observability: one line per member so the journal shows
// the fallback firing and how many tags each repo contributed (the error
// lines above only fire on failure — a clean run was otherwise silent).
eprintln!("git-fallback: {repo_name} → {} tags", nodes.len());
if !nodes.is_empty() {
nodes.sort_by_key(|n| n.timestamp);
lanes.push(Lane {
repo: repo_name,
nodes,
});
}
}
lanes.sort_by(|a, b| a.repo.cmp(&b.repo));
eprintln!(
"git-fallback: {git_dir:?} → {} repo lane(s) with tags",
lanes.len()
);
lanes
}
pub fn load_timeline(warehouse_root: &Path, workspace_name: &str) -> Result<Timeline> {
// Read-only + lock-tolerant: the timeline is a pure read, so don't take the
// exclusive redb writer lock — fall back to a copied snapshot when a writer
// (a live nornir-server, or a concurrent test) holds it. (open_read_only)
let wh = IcebergWarehouse::open_read_only(warehouse_root)?;
build_timeline(&wh, workspace_name, Some(warehouse_root))
}
/// Core timeline builder over an already-open warehouse handle. Shared by the
/// local loader (above) and the server's `Viz.Timeline` RPC, so the viz can run
/// either embedded (own warehouse) or as a thin gRPC client. `bench_fallback_dir`
/// is the warehouse root used only to synthesize empty lanes for repos that have
/// bench rows but no recorded release; pass `None` when the path isn't known
/// (e.g. server-side) — the fallback is then skipped (a cosmetic nicety only).
pub fn build_timeline(
wh: &IcebergWarehouse,
workspace_name: &str,
bench_fallback_dir: Option<&Path>,
) -> Result<Timeline> {
let reports: Vec<ReleaseReport> =
wh.block_on(query_release_history(wh, workspace_name, None))?;
let snapshots_vec = wh.block_on(query_dep_graph_snapshots(wh, workspace_name, None))?;
let mut snapshots: BTreeMap<Uuid, DepGraphSnapshot> = BTreeMap::new();
let latest_snapshot = snapshots_vec.last().cloned();
for s in snapshots_vec {
snapshots.insert(s.snapshot_id, s);
}
let mut by_repo: BTreeMap<String, Vec<LaneNode>> = BTreeMap::new();
let mut release_order: Vec<Uuid> = Vec::new();
let mut release_snapshot: BTreeMap<Uuid, Uuid> = BTreeMap::new();
let mut seen: BTreeSet<Uuid> = BTreeSet::new();
for rep in &reports {
if seen.insert(rep.release_id) {
release_order.push(rep.release_id);
release_snapshot.insert(rep.release_id, rep.dep_graph_snapshot_id);
}
// Real release timestamp. The release_lineage table persists one
// (`query_release_history` even reads it into its grouping key), but
// `ReleaseReport` has no `timestamp` field to carry it out — so the viz
// can't see it directly. Each release pins the dep-graph snapshot
// captured during the SAME release run, and that `DepGraphSnapshot`
// *does* reach us with a real `timestamp`; use it as the release's
// recorded time. Falls back to `Utc::now()` only for pre-1.x rows whose
// snapshot id is `Uuid::nil()` (no snapshot was captured).
//
// TODO: the exact fix is to add a `timestamp: DateTime<Utc>` field to
// `ReleaseReport` (src/release/pipeline.rs) populated from the lineage
// `tss` column already read at query time, then use it here. That file
// is out of scope for this viz-only change.
let when = snapshots
.get(&rep.dep_graph_snapshot_id)
.map(|s| s.timestamp)
.unwrap_or_else(Utc::now);
for r in &rep.repos {
by_repo.entry(r.repo.clone()).or_default().push(LaneNode {
release_id: rep.release_id,
timestamp: when,
sha: r.git.sha.clone(),
branch: r.git.branch.clone(),
dirty: r.git.dirty,
gate_status: r.gate_status.clone(),
tests_passed: r.tests_passed,
tests_failed: r.tests_failed,
published_versions: r.published_versions.clone(),
});
}
}
let mut lanes: Vec<Lane> = by_repo
.into_iter()
.map(|(repo, mut nodes)| {
nodes.sort_by_key(|n| n.timestamp);
Lane { repo, nodes }
})
.collect();
// Git-history fallback: when a workspace has NO recorded releases (empty
// warehouse / never republished), read the WHOLE GIT HISTORY of each
// monitored member — its tags become lane nodes — so the timeline shows the
// repo's real release history instead of an empty placeholder. The monitor's
// git clones live in `../git/<member>/` beside the warehouse root.
if lanes.is_empty() {
if let Some(git_dir) = bench_fallback_dir
.and_then(|root| root.parent())
.map(|p| p.join("git"))
{
lanes = git_history_lanes(&git_dir);
// Each node's x-column is its index in `release_order`; without this
// every fallback node resolves to col 0 and they pile up invisibly at
// the far left. Build the order chronologically across all lanes.
let mut chron: Vec<(DateTime<Utc>, Uuid)> = lanes
.iter()
.flat_map(|l| l.nodes.iter().map(|n| (n.timestamp, n.release_id)))
.collect();
chron.sort_by_key(|(t, _)| *t);
release_order = chron.into_iter().map(|(_, id)| id).collect();
}
}
// Bench-only fallback: if there were STILL no releases, scan the
// `bench_runs/repo=*/` partition directories and synthesize empty lanes so
// the TimeTravel benchmarks reel still has data.
if lanes.is_empty() {
if let Some(rd) = bench_fallback_dir
.map(|root| root.join("bench_runs"))
.and_then(|bench_dir| std::fs::read_dir(bench_dir).ok())
{
for entry in rd.flatten() {
let name = entry.file_name().to_string_lossy().to_string();
if let Some(repo) = name.strip_prefix("repo=") {
lanes.push(Lane {
repo: repo.to_string(),
nodes: Vec::new(),
});
}
}
lanes.sort_by(|a, b| a.repo.cmp(&b.repo));
}
}
// Load bench history per repo for the TimeTravel reels.
let mut bench_history: BTreeMap<String, BenchHistory> = BTreeMap::new();
for lane in &lanes {
let runs = wh
.query_bench_runs(&BenchFilter::for_repo(&lane.repo))
.unwrap_or_default();
let mut points: Vec<BenchPoint> =
runs.into_iter().map(|run| bench_point_from(&run)).collect();
points.sort_by_key(|p| p.timestamp);
bench_history.insert(
lane.repo.clone(),
BenchHistory {
repo: lane.repo.clone(),
points,
},
);
}
Ok(Timeline {
workspace_name: workspace_name.to_string(),
lanes,
release_order,
release_snapshot,
snapshots,
latest_snapshot,
bench_history,
})
}
fn bench_point_from(run: &BenchRun) -> BenchPoint {
let ts = run
.timestamp
.as_deref()
.and_then(|s| DateTime::parse_from_rfc3339(s).ok())
.map(|d| d.with_timezone(&Utc))
.or_else(|| {
chrono::NaiveDate::parse_from_str(&run.date, "%Y-%m-%d")
.ok()
.and_then(|d| d.and_hms_opt(0, 0, 0))
.map(|n| DateTime::<Utc>::from_naive_utc_and_offset(n, Utc))
})
.unwrap_or_else(Utc::now);
let mut metrics: Vec<(String, f64)> = Vec::new();
for r in &run.results {
for (k, v) in &r.metrics {
if let Some(f) = v.as_f64() {
metrics.push((format!("{}.{}", r.name, k), f));
} else if let Some(i) = v.as_i64() {
metrics.push((format!("{}.{}", r.name, k), i as f64));
}
}
}
let (primary_metric_name, primary_metric_value) = metrics
.iter()
.max_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal))
.cloned()
.unwrap_or_else(|| (String::from("(none)"), 0.0));
BenchPoint {
timestamp: ts,
primary_metric_name,
primary_metric_value,
metrics,
version: run.version.clone(),
machine: run.machine.clone(),
}
}
#[cfg(test)]
mod tests {
use super::*;
// The `Timeline` JSON round-trip (the `Viz.Timeline` wire contract) moved with
// the type to `nornir_wire::view::timeline::tests::timeline_json_roundtrip` —
// the leaf that now owns the shape proves it. This module keeps the PRODUCER
// test (`git_history_lanes`), which is nornir-only (reads a real git repo).
/// The git-history fallback (`git_history_lanes`) must turn a member repo's
/// **tags** into lane nodes — the path that fills the timeline for a
/// workspace with no recorded releases. Builds a real temp git repo with two
/// version tags and asserts the lane, node count, and parsed versions.
#[test]
fn git_history_lanes_reads_member_tags() {
use std::process::Command;
let dir = tempfile::tempdir().unwrap();
let git_dir = dir.path().join("git");
let repo = git_dir.join("myrepo");
std::fs::create_dir_all(&repo).unwrap();
let git = |args: &[&str]| {
let ok = Command::new("git")
.args(args)
.current_dir(&repo)
.env("GIT_AUTHOR_NAME", "t")
.env("GIT_AUTHOR_EMAIL", "t@e")
.env("GIT_COMMITTER_NAME", "t")
.env("GIT_COMMITTER_EMAIL", "t@e")
.output()
.expect("run git")
.status
.success();
assert!(ok, "git {args:?} failed");
};
git(&["init", "-q", "-b", "main"]);
std::fs::write(repo.join("f"), "1").unwrap();
git(&["add", "."]);
git(&["commit", "-q", "-m", "one"]);
git(&["tag", "v0.1.0"]);
std::fs::write(repo.join("f"), "2").unwrap();
git(&["add", "."]);
git(&["commit", "-q", "-m", "two"]);
git(&["tag", "myrepo-v0.2.0"]);
let lanes = git_history_lanes(&git_dir);
assert_eq!(lanes.len(), 1, "one member repo → one lane");
assert_eq!(lanes[0].repo, "myrepo");
assert_eq!(lanes[0].nodes.len(), 2, "two tags → two nodes");
let versions: Vec<String> = lanes[0]
.nodes
.iter()
.flat_map(|n| n.published_versions.iter().map(|(_, v)| v.clone()))
.collect();
assert!(
versions.contains(&"0.1.0".to_string()),
"v0.1.0 parsed: {versions:?}"
);
assert!(
versions.contains(&"0.2.0".to_string()),
"myrepo-v0.2.0 parsed: {versions:?}"
);
assert!(
lanes[0].nodes.iter().all(|n| n.sha.len() >= 7),
"each node has a sha"
);
}
}
// ─── the live-run event stream ──────────────────────────────────────────────
//
// LIVES HERE, not in `live.rs`, because two consumers need it and only one of
// them is a GUI: the egui live tab renders these, and the gRPC thin client
// (`super::remote`) deserializes them off `Viz.Timeline`. `live.rs` links
// eframe/egui, so a server build cannot see it — and the thin client is exactly
// what a server-featured build of the CLI wants. A pure serde enum has no
// business gating a wire client on a window toolkit.
#[derive(Debug, Clone, Deserialize)]
#[serde(tag = "kind")]
pub enum LiveEvent {
#[serde(rename = "run_start")]
RunStart { run_id: String, workspace: String },
#[serde(rename = "repo_start")]
RepoStart { repo: String, sha: String },
#[serde(rename = "phase_start")]
PhaseStart { repo: String, phase: String },
#[serde(rename = "phase_end")]
PhaseEnd {
repo: String,
phase: String,
ok: bool,
duration_ms: u64,
},
#[serde(rename = "binary_start")]
BinaryStart { repo: String, binary: String },
#[serde(rename = "test_pass")]
TestPass {
repo: String,
binary: String,
name: String,
},
#[serde(rename = "test_fail")]
TestFail {
repo: String,
binary: String,
name: String,
},
#[serde(rename = "binary_done")]
BinaryDone {
repo: String,
binary: String,
passed: u32,
failed: u32,
},
#[serde(rename = "repo_end")]
RepoEnd { repo: String, ok: bool },
#[serde(rename = "run_end")]
RunEnd { run_id: String, ok: bool },
}