diff --git a/Cargo.toml b/Cargo.toml index 356ea49..7afded9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -111,6 +111,11 @@ name = "question_lint" path = "src/bin/question_lint.rs" required-features = ["ssr"] +[[bin]] +name = "needs_replay" +path = "src/bin/needs_replay.rs" +required-features = ["ssr"] + [profile.wasm-release] inherits = "release" opt-level = 'z' diff --git a/src/bin/needs_replay.rs b/src/bin/needs_replay.rs new file mode 100644 index 0000000..d772cee --- /dev/null +++ b/src/bin/needs_replay.rs @@ -0,0 +1,107 @@ +//! Makes a simulation's cases real: every case becomes an aggregate in +//! its bucket, created and moved by the same code the running site +//! uses - `answers::store_answer` for the submission, +//! `aggregates::transition` for each desk move, against a real +//! JetStream with its event log and its refusal of undeclared moves. +//! Then the buckets are read back by replaying each record's events, +//! and reported per need against `done_when` and `success`. +//! +//! The report half runs on its own (`--report-only`) against any +//! JetStream, which is how the same scorecard is taken from a live +//! site's buckets. +//! +//! Point it at a throwaway NATS, never a site's own: it writes. +#![cfg(feature = "ssr")] + +use std::collections::{BTreeMap, HashMap}; + +use futures::StreamExt; +use portal::events::store::{ensure_stream, load_events}; +use portal::local::{load_aggregates_from_dir, load_from_dir, load_needs_from_dir}; +use portal::{aggregates, answers, content, needs}; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let argv: Vec = std::env::args().skip(1).collect(); + let value = |flag: &str| argv.iter().position(|a| a == flag).and_then(|i| argv.get(i + 1)).cloned(); + let usage = "usage: needs_replay --path --nats (--cases | --report-only)"; + let dir = value("--path").ok_or_else(|| anyhow::anyhow!(usage))?; + let nats_url = value("--nats").ok_or_else(|| anyhow::anyhow!(usage))?; + let cases_path = value("--cases"); + if cases_path.is_none() && !argv.iter().any(|a| a == "--report-only") { + anyhow::bail!(usage); + } + + let questions = load_from_dir(&dir)?; + let schemas = content::with_builtin_aggregates(load_aggregates_from_dir(&dir)?); + let file = needs::parse(&load_needs_from_dir(&dir)?.ok_or_else(|| anyhow::anyhow!("the repo has no needs.yaml"))?)?; + let js = async_nats::jetstream::new(async_nats::connect(&nats_url).await?); + // The same call the site makes at boot. + ensure_stream(&js).await?; + + if let Some(path) = cases_path { + let raw = std::fs::read_to_string(&path).map_err(|e| anyhow::anyhow!("reading {path}: {e}"))?; + let (mut created, mut moved, mut refused) = (0, 0, 0); + for (n, line) in raw.lines().filter(|l| !l.trim().is_empty()).enumerate() { + let case: serde_json::Value = serde_json::from_str(line)?; + let Some(need) = file.needs.iter().find(|need| case["need"] == need.id.as_str()) else { continue }; + let Some(schema) = schemas.get(&need.lands_in) else { continue }; + // Only a case that got as far as submitting leaves a record. + let states: Vec<&str> = case["states"].as_array().into_iter().flatten().filter_map(|s| s.as_str()).collect(); + if states.is_empty() { + continue; + } + let step = needs::path_for(&questions, need).and_then(|p| p.last().cloned()); + let (page, alternative) = step.map(|s| (s.page, s.alternative)).unwrap_or_default(); + let id = format!("sim-{}-{}-{n}", case["persona"].as_str().unwrap_or("x"), case["run"]); + let now = chrono::Utc::now().timestamp_millis(); + answers::store_answer(&js, &schemas, &need.lands_in, id.clone(), &page, &alternative, &case["record"], now).await?; + created += 1; + for target in &states[1..] { + match aggregates::transition(&js, schema, &id, target, serde_json::json!({"by": "simulation"}), now).await { + Ok(_) => moved += 1, + Err(e) => { + refused += 1; + eprintln!("REFUSED by the engine: {id} -> {target}: {e:?}"); + break; + } + } + } + } + println!("{created} aggregate(s) created, {moved} transition(s) accepted, {refused} refused by the engine\n"); + } + + // Read every bucket back the way the engine knows it: by replaying + // each record's own event log. + let mut by_bucket: HashMap> = HashMap::new(); + for bucket in file.needs.iter().map(|n| n.lands_in.clone()).collect::>() { + let (Some(schema), Ok(store)) = (schemas.get(&bucket), js.get_key_value(&bucket).await) else { continue }; + let mut keys = store.keys().await?; + while let Some(Ok(id)) = keys.next().await { + let events = load_events(&js, &bucket, &id).await?; + if let Some(record) = aggregates::replay(schema, &id, &events) { + *by_bucket.entry(bucket.clone()).or_default().entry(record.state).or_default() += 1; + } + } + } + println!("{:<18} {:>8} {:>9} {:>9} {}", "bucket", "records", "finished", "success", "by state"); + let mut buckets: Vec<_> = by_bucket.iter().collect(); + buckets.sort(); + for (bucket, states) in buckets { + let needs_here: Vec<&needs::Need> = file.needs.iter().filter(|n| &n.lands_in == bucket).collect(); + let is = |pick: fn(&needs::Need) -> &Vec, state: &String| needs_here.iter().any(|n| pick(n).contains(state)); + let total: usize = states.values().sum(); + let finished: usize = states.iter().filter(|(s, _)| is(|n| &n.done_when, s)).map(|(_, c)| c).sum(); + let success: usize = states.iter().filter(|(s, _)| is(|n| &n.success, s)).map(|(_, c)| c).sum(); + let pct = |n: usize| if total == 0 { "-".to_string() } else { format!("{:.0}%", n as f64 * 100.0 / total as f64) }; + println!( + "{:<18} {:>8} {:>9} {:>9} {}", + bucket, + total, + pct(finished), + pct(success), + states.iter().map(|(s, c)| format!("{s} {c}")).collect::>().join(", ") + ); + } + Ok(()) +} diff --git a/src/bin/question_lint.rs b/src/bin/question_lint.rs index 73ce1f4..dedfdb4 100644 --- a/src/bin/question_lint.rs +++ b/src/bin/question_lint.rs @@ -9,6 +9,7 @@ //! static binary," not "build a Rust workspace." #![cfg(feature = "ssr")] +use portal::local::{load_aggregates_from_dir, load_from_dir, load_needs_from_dir, load_site_from_dir}; use portal::{aggregates, content, needs}; #[tokio::main] @@ -108,97 +109,6 @@ async fn main() -> anyhow::Result<()> { } } -/// The offline counterpart to `content::load_site_from_gitea` - -/// `site.yaml` lives at the repo root like `aggregates.yaml`, and is -/// just as optional locally: missing means default branding. -fn load_site_from_dir(dir: &str) -> anyhow::Result { - let site_path = std::path::Path::new(dir) - .parent() - .unwrap_or_else(|| std::path::Path::new(".")) - .join("site.yaml"); - if !site_path.exists() { - return Ok(content::SiteConfig::default()); - } - let raw = std::fs::read_to_string(&site_path) - .map_err(|e| anyhow::anyhow!("reading {}: {e}", site_path.display()))?; - serde_yaml::from_str(&raw).map_err(|e| anyhow::anyhow!("parsing {}: {e}", site_path.display())) -} - -/// The offline counterpart to `content::load_aggregates_from_gitea` - -/// `aggregates.yaml` lives at the repo root, one level up from the -/// pages directory `--path` names, so `dir`'s parent is where it's -/// looked for. Missing entirely is not an error here (unlike -/// `--repo` mode, where `aggregates.yaml` is required) - a local -/// checkout being linted may not have one, and every declared -/// transition still gets checked against whatever *is* found; an -/// empty map just means nothing is checked. -fn load_aggregates_from_dir( - dir: &str, -) -> anyhow::Result> { - let aggregates_path = std::path::Path::new(dir) - .parent() - .unwrap_or_else(|| std::path::Path::new(".")) - .join("aggregates.yaml"); - if !aggregates_path.exists() { - return Ok(std::collections::HashMap::new()); - } - let raw = std::fs::read_to_string(&aggregates_path) - .map_err(|e| anyhow::anyhow!("reading {}: {e}", aggregates_path.display()))?; - aggregates::parse_aggregates_yaml(&raw) -} - -/// The offline counterpart to `content::load_questions_from_gitea` - -/// the same recursive tree walk and `content::build_questions` -/// pipeline (derived ids, relative refs, sections, followup -/// inference), just reading a local checkout instead of Gitea's API, -/// for linting a branch that hasn't been pushed yet. -fn load_from_dir( - dir: &str, -) -> anyhow::Result> { - let base = std::path::Path::new(dir); - let mut files = Vec::new(); - collect_yaml(base, base, &mut files)?; - content::build_questions(&files) -} - -fn collect_yaml( - base: &std::path::Path, - dir: &std::path::Path, - out: &mut Vec<(String, String)>, -) -> anyhow::Result<()> { - for entry in std::fs::read_dir(dir)? { - let path = entry?.path(); - if path.is_dir() { - collect_yaml(base, &path, out)?; - } else if path.extension().and_then(|e| e.to_str()) == Some("yaml") { - let rel = path - .strip_prefix(base) - .expect("walked paths sit under their base") - .to_string_lossy() - .replace('\\', "/"); - let raw = std::fs::read_to_string(&path) - .map_err(|e| anyhow::anyhow!("reading {}: {e}", path.display()))?; - out.push((rel, raw)); - } - } - Ok(()) -} - -/// The offline counterpart to `content::load_needs_from_gitea` - -/// `needs.yaml` sits at the repo root beside `aggregates.yaml`. -fn load_needs_from_dir(dir: &str) -> anyhow::Result> { - let needs_path = std::path::Path::new(dir) - .parent() - .unwrap_or_else(|| std::path::Path::new(".")) - .join("needs.yaml"); - if !needs_path.exists() { - return Ok(None); - } - std::fs::read_to_string(&needs_path) - .map(Some) - .map_err(|e| anyhow::anyhow!("reading {}: {e}", needs_path.display())) -} - /// The business-needs pass (see `portal::needs`): structure always, /// then the wording tasks written out and/or an engine's answers /// scored, when asked. A repo without `needs.yaml` skips all of it. diff --git a/src/lib.rs b/src/lib.rs index 38db8da..9bb6325 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -16,6 +16,9 @@ pub mod announce; #[cfg(feature = "ssr")] pub mod needs; +#[cfg(feature = "ssr")] +pub mod local; + #[cfg(feature = "ssr")] pub mod server; diff --git a/src/local.rs b/src/local.rs new file mode 100644 index 0000000..e60b9d5 --- /dev/null +++ b/src/local.rs @@ -0,0 +1,100 @@ +//! Loading a content repo from a local checkout - the offline +//! counterparts to the Gitea loaders in `content`, for tools that lint +//! or replay a branch that has not been pushed (question_lint, +//! needs_replay). `dir` is always the pages directory (`questions/`); +//! `site.yaml`, `aggregates.yaml` and `needs.yaml` sit one level up. +#![cfg(feature = "ssr")] + +use crate::{aggregates, content}; + +/// The offline counterpart to `content::load_site_from_gitea` - +/// `site.yaml` lives at the repo root like `aggregates.yaml`, and is +/// just as optional locally: missing means default branding. +pub fn load_site_from_dir(dir: &str) -> anyhow::Result { + let site_path = std::path::Path::new(dir) + .parent() + .unwrap_or_else(|| std::path::Path::new(".")) + .join("site.yaml"); + if !site_path.exists() { + return Ok(content::SiteConfig::default()); + } + let raw = std::fs::read_to_string(&site_path) + .map_err(|e| anyhow::anyhow!("reading {}: {e}", site_path.display()))?; + serde_yaml::from_str(&raw).map_err(|e| anyhow::anyhow!("parsing {}: {e}", site_path.display())) +} + +/// The offline counterpart to `content::load_aggregates_from_gitea` - +/// `aggregates.yaml` lives at the repo root, one level up from the +/// pages directory `--path` names, so `dir`'s parent is where it's +/// looked for. Missing entirely is not an error here (unlike +/// `--repo` mode, where `aggregates.yaml` is required) - a local +/// checkout being linted may not have one, and every declared +/// transition still gets checked against whatever *is* found; an +/// empty map just means nothing is checked. +pub fn load_aggregates_from_dir( + dir: &str, +) -> anyhow::Result> { + let aggregates_path = std::path::Path::new(dir) + .parent() + .unwrap_or_else(|| std::path::Path::new(".")) + .join("aggregates.yaml"); + if !aggregates_path.exists() { + return Ok(std::collections::HashMap::new()); + } + let raw = std::fs::read_to_string(&aggregates_path) + .map_err(|e| anyhow::anyhow!("reading {}: {e}", aggregates_path.display()))?; + aggregates::parse_aggregates_yaml(&raw) +} + +/// The offline counterpart to `content::load_questions_from_gitea` - +/// the same recursive tree walk and `content::build_questions` +/// pipeline (derived ids, relative refs, sections, followup +/// inference), just reading a local checkout instead of Gitea's API, +/// for linting a branch that hasn't been pushed yet. +pub fn load_from_dir( + dir: &str, +) -> anyhow::Result> { + let base = std::path::Path::new(dir); + let mut files = Vec::new(); + collect_yaml(base, base, &mut files)?; + content::build_questions(&files) +} + +fn collect_yaml( + base: &std::path::Path, + dir: &std::path::Path, + out: &mut Vec<(String, String)>, +) -> anyhow::Result<()> { + for entry in std::fs::read_dir(dir)? { + let path = entry?.path(); + if path.is_dir() { + collect_yaml(base, &path, out)?; + } else if path.extension().and_then(|e| e.to_str()) == Some("yaml") { + let rel = path + .strip_prefix(base) + .expect("walked paths sit under their base") + .to_string_lossy() + .replace('\\', "/"); + let raw = std::fs::read_to_string(&path) + .map_err(|e| anyhow::anyhow!("reading {}: {e}", path.display()))?; + out.push((rel, raw)); + } + } + Ok(()) +} + +/// The offline counterpart to `content::load_needs_from_gitea` - +/// `needs.yaml` sits at the repo root beside `aggregates.yaml`. +pub fn load_needs_from_dir(dir: &str) -> anyhow::Result> { + let needs_path = std::path::Path::new(dir) + .parent() + .unwrap_or_else(|| std::path::Path::new(".")) + .join("needs.yaml"); + if !needs_path.exists() { + return Ok(None); + } + std::fs::read_to_string(&needs_path) + .map(Some) + .map_err(|e| anyhow::anyhow!("reading {}: {e}", needs_path.display())) +} +