portal/src/lint/replay.rs

//! 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.

use std::collections::{BTreeMap, HashMap};

use futures::StreamExt;
use crate::events::store::{ensure_stream, load_events};
use crate::lint::local::{load_aggregates_from_dir, load_from_dir, load_needs_from_dir};
use crate::lint::needs;
use crate::{aggregates, answers, content};

pub async fn run(argv: Vec<String>) -> anyhow::Result<()> {
    let value = |flag: &str| argv.iter().position(|a| a == flag).and_then(|i| argv.get(i + 1)).cloned();
    let usage = "usage: iris replay --path <questions dir> --nats <url> (--cases <cases.jsonl> | --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, 0).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<String, BTreeMap<String, usize>> = HashMap::new();
    for bucket in file.needs.iter().map(|n| n.lands_in.clone()).collect::<std::collections::BTreeSet<_>>() {
        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<String>, 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::<Vec<_>>().join(", ")
        );
    }
    Ok(())
}