nornir 0.4.15

//! Funnel **demo** generator + nuke — the E1 "funnel demo" deliverable.
//!
//! Two halves that mirror each other:
//!
//!   1. **On-disk fake repos.** [`run_demo`] materialises `--size N` tiny but
//!      *real* git repos near the `git/` root, each a `Cargo.toml` +
//!      `src/lib.rs`, wired into a **layered acyclic dependency graph** via
//!      path-deps on lower-indexed siblings (so `funnel_fake_3` depends on
//!      `funnel_fake_1` and `funnel_fake_2`, never the reverse → guaranteed
//!      acyclic). Every created path is recorded in a **manifest**
//!      `git/.nornir-funnel-demo.json` so [`nuke_disk`] deletes *exactly* what
//!      the demo made — never a heuristic `rm`.
//!
//!   2. **In-warehouse funnel DAG.** The same run injects a funnel idea→plan
//!      whose node DAG **mirrors** the on-disk repo dep graph (one node per
//!      fake repo, one edge per path-dep). The idea is tagged
//!      `source = "funnel:demo"` so reruns are **idempotent** (an existing demo
//!      idea is detected and the inject is skipped).
//!
//! Nuke is **disk-only**: it removes the fake repos + manifest but leaves the
//! warehouse funnel data untouched, so the plan still shows up in `funnel show`
//! and the viz Funnel tab after a nuke (the event log is append-only history).
//!
//! ### Determinism
//! [`inject_dep_graph`] builds the funnel DAG from an explicit `seed` with a
//! tiny splitmix64 PRNG, so the *expected* node/edge/ready set is computable in
//! tests (golden, data-only). Edges only ever go **lower-layer → higher-layer**,
//! making the graph acyclic by construction.

use std::path::{Path, PathBuf};

use anyhow::{Context, Result};
use chrono::Utc;
use serde::{Deserialize, Serialize};

use super::event::{Event, PlanStatus};
use super::ids::{IdeaId, NodeId, PlanId};
use super::store::Store;

/// The `source` tag stamped on the demo idea. Reruns key off this so a second
/// `run_demo`/`inject_demo` is a no-op instead of a duplicate plan.
pub const DEMO_SOURCE: &str = "funnel:demo";

/// Manifest filename written into the git root listing exactly what the demo
/// created on disk, so nuke is precise (never a heuristic delete).
pub const MANIFEST_NAME: &str = ".nornir-funnel-demo.json";

// ─── deterministic PRNG ─────────────────────────────────────────────────────

/// splitmix64 — a tiny, deterministic, dependency-free PRNG. Same `seed` →
/// same stream, so the injected DAG (and therefore the expected golden state)
/// is fully reproducible.
struct SplitMix64(u64);

impl SplitMix64 {
    fn new(seed: u64) -> Self {
        Self(seed)
    }
    fn next_u64(&mut self) -> u64 {
        self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15);
        let mut z = self.0;
        z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
        z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
        z ^ (z >> 31)
    }
    /// Uniform-ish `0..n` (n>0).
    fn below(&mut self, n: usize) -> usize {
        (self.next_u64() % (n as u64)) as usize
    }
    /// Coin flip biased by `num/den`.
    fn chance(&mut self, num: u64, den: u64) -> bool {
        self.next_u64() % den < num
    }
}

// ─── manifest ───────────────────────────────────────────────────────────────

/// What the demo created on disk, written to `git/.nornir-funnel-demo.json`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DemoManifest {
    /// Schema marker.
    pub kind: String,
    /// When the demo was generated.
    pub created_at: chrono::DateTime<Utc>,
    /// `--size N`.
    pub size: usize,
    /// Absolute paths of every fake-repo dir the demo created (the only paths
    /// nuke is allowed to delete).
    pub repos: Vec<PathBuf>,
    /// The container dir created for `size > 1` (e.g. `git/funnel3_fake_projects`),
    /// removed by nuke once empty. `None` for the `size == 1` single-repo layout.
    pub container: Option<PathBuf>,
}

impl DemoManifest {
    fn path(git_root: &Path) -> PathBuf {
        git_root.join(MANIFEST_NAME)
    }
    fn load(git_root: &Path) -> Option<Self> {
        let p = Self::path(git_root);
        let txt = std::fs::read_to_string(p).ok()?;
        serde_json::from_str(&txt).ok()
    }
    fn save(&self, git_root: &Path) -> Result<()> {
        let p = Self::path(git_root);
        let txt = serde_json::to_string_pretty(self).context("serialize demo manifest")?;
        std::fs::write(&p, txt).with_context(|| format!("write manifest {}", p.display()))?;
        Ok(())
    }
}

// ─── git-root resolution ────────────────────────────────────────────────────

/// Resolve the `git/` root the fake repos live under.
///
/// Precedence:
///   1. the warehouse `../git/` sibling convention — the warehouse root is
///      `<ws>/.nornir/warehouse`, so its great-grandparent's `git/` is the
///      monorepo git root (matches the timeline's `../git/<member>` lookup);
///   2. fallback: `<workspace_root>/../git/`.
///
/// The funnel root passed to [`run_demo`] is the warehouse root, so we walk up
/// from it. We accept either the `.nornir/warehouse` shape or a bare dir.
pub fn resolve_git_root(funnel_root: &Path) -> PathBuf {
    // <ws>/.nornir/warehouse → walk to <ws>, then sibling git/.
    // warehouse → .nornir → <ws>
    let ws = funnel_root
        .parent() // .nornir
        .and_then(|p| p.parent()) // <ws>
        .unwrap_or(funnel_root);
    // Prefer <ws>/../git (the monorepo sibling), else <ws>/git.
    if let Some(parent) = ws.parent() {
        let sib = parent.join("git");
        if sib.exists() {
            return sib;
        }
        return sib; // create-on-demand: still the sibling convention
    }
    ws.join("git")
}

// ─── on-disk repo names ─────────────────────────────────────────────────────

/// Directory each fake repo lives in, for `size` and 1-based index `i`.
///
///   * `size == 1` → `<git>/funnel_fake/funnel1_fake_project/`
///   * `size  > 1` → `<git>/funnel{N}_fake_projects/funnel_fake_{i}/`
fn repo_dir(git_root: &Path, size: usize, i: usize) -> PathBuf {
    if size == 1 {
        git_root.join("funnel_fake").join("funnel1_fake_project")
    } else {
        git_root
            .join(format!("funnel{size}_fake_projects"))
            .join(format!("funnel_fake_{i}"))
    }
}

/// The container dir (parent of all repos) for nuke cleanup.
fn container_dir(git_root: &Path, size: usize) -> PathBuf {
    if size == 1 {
        git_root.join("funnel_fake")
    } else {
        git_root.join(format!("funnel{size}_fake_projects"))
    }
}

/// Crate name for the fake repo at 1-based index `i`.
fn crate_name(i: usize) -> String {
    format!("funnel_fake_{i}")
}

// ─── the funnel DAG injector (the data-plane half) ──────────────────────────

/// One plan node in the injected demo DAG, paired with the edges into it.
#[derive(Debug, Clone)]
pub struct DemoNodeSpec {
    pub node_id: NodeId,
    pub kind: String,
    pub targets: Vec<String>,
    /// node ids this node depends on (lower-layer → this).
    pub deps: Vec<NodeId>,
    pub layer: usize,
}

/// Build the **layered acyclic** demo DAG spec for `plans` plans, each with
/// `nodes_per_plan` nodes arranged into `layers` layers, deterministically from
/// `seed`. Edges only go lower-layer → higher-layer (acyclic by construction).
///
/// Returns, per plan, the ordered list of node specs. The caller records the
/// matching events through a [`Store`]; tests recompute the expected golden set
/// from the same specs.
pub fn build_dep_graph_spec(
    plans: usize,
    nodes_per_plan: usize,
    layers: usize,
    seed: u64,
) -> Vec<Vec<DemoNodeSpec>> {
    let mut rng = SplitMix64::new(seed);
    let layers = layers.max(1);
    let mut out = Vec::with_capacity(plans);
    let mut next_node = 1u64;

    for _plan in 0..plans {
        let n = nodes_per_plan.max(1);
        // Assign each node to a layer: spread round-robin so every layer is
        // populated, then nudge a few by the PRNG for variety (still monotone:
        // we only ever assign a layer index, edges enforce acyclicity).
        let mut node_layer = Vec::with_capacity(n);
        for k in 0..n {
            let base = (k * layers) / n; // monotone fill 0..layers
            let jitter = if layers > 1 && rng.chance(1, 4) {
                // occasionally push down one layer (never up past its base+1)
                (base + 1).min(layers - 1)
            } else {
                base
            };
            node_layer.push(jitter);
        }
        // Ensure at least one layer-0 root.
        if !node_layer.iter().any(|&l| l == 0) {
            node_layer[0] = 0;
        }

        // Allocate node ids in declaration order.
        let ids: Vec<NodeId> = (0..n)
            .map(|_| {
                let id = NodeId::seq(next_node);
                next_node += 1;
                id
            })
            .collect();

        // Build edges: each non-root node gets 1..=2 deps chosen from STRICTLY
        // lower layers (deterministic). If no lower-layer node exists, it stays
        // a root.
        let mut specs = Vec::with_capacity(n);
        for k in 0..n {
            let layer = node_layer[k];
            let lower: Vec<usize> =
                (0..n).filter(|&j| node_layer[j] < layer).collect();
            let mut deps = Vec::new();
            if !lower.is_empty() {
                let want = 1 + rng.below(2.min(lower.len())); // 1 or 2 deps
                let mut chosen = std::collections::BTreeSet::new();
                for _ in 0..want {
                    chosen.insert(lower[rng.below(lower.len())]);
                }
                deps = chosen.into_iter().map(|j| ids[j].clone()).collect();
            }
            specs.push(DemoNodeSpec {
                node_id: ids[k].clone(),
                kind: "code:write".to_string(),
                targets: vec![format!("{}/src/lib.rs", crate_name(k + 1))],
                deps,
                layer,
            });
        }
        out.push(specs);
    }
    out
}

/// Inject the layered demo DAG into `store` as one idea + `plans` plans, each
/// carrying its `nodes_per_plan`-node DAG. Idempotent: if a `source =
/// "funnel:demo"` idea already exists, returns `Ok(None)` without writing.
///
/// On success returns `Some((idea_id, plan_ids))`.
pub fn inject_dep_graph(
    store: &mut Store,
    plans: usize,
    nodes_per_plan: usize,
    layers: usize,
    seed: u64,
) -> Result<Option<(IdeaId, Vec<PlanId>)>> {
    // Idempotency: already injected?
    if store
        .funnel
        .ideas
        .values()
        .any(|i| i.source == DEMO_SOURCE)
    {
        return Ok(None);
    }

    // Strictly increasing timestamps so replay order is deterministic (the
    // Iceberg scan order is not stable; replay sorts/processes by ts).
    let base = Utc::now();
    let mut tick = 0i64;
    let mut at = || {
        tick += 1;
        base + chrono::Duration::microseconds(tick)
    };

    let idea_id = IdeaId::seq(store.funnel.next_idea.max(1));
    store.record(Event::IdeaSubmitted {
        id: idea_id.clone(),
        source: DEMO_SOURCE.to_string(),
        text: format!(
            "funnel demo: {plans} plan(s) × {nodes_per_plan} nodes × {layers} layers (seed {seed})"
        ),
        refs: vec![],
        ts: at(),
    })?;

    let specs = build_dep_graph_spec(plans, nodes_per_plan, layers, seed);
    let mut plan_ids = Vec::with_capacity(plans);

    for (pi, plan_specs) in specs.iter().enumerate() {
        let plan_id = PlanId::seq(store.funnel.next_plan.max(1));
        store.record(Event::PlanCreated {
            id: plan_id.clone(),
            idea_id: idea_id.clone(),
            summary: format!("demo plan {} (fake dep graph)", pi + 1),
            planner: DEMO_SOURCE.to_string(),
            ts: at(),
        })?;
        store.record(Event::PlanStatusChanged {
            plan_id: plan_id.clone(),
            status: PlanStatus::Active,
            why: None,
            ts: at(),
        })?;

        // Nodes first (so edges can reference them), then edges.
        for spec in plan_specs {
            store.record(Event::NodeAdded {
                plan_id: plan_id.clone(),
                node_id: spec.node_id.clone(),
                kind: spec.kind.clone(),
                params: serde_json::Map::new(),
                targets: spec.targets.clone(),
                prompt_excerpt: None,
                ts: at(),
            })?;
        }
        for spec in plan_specs {
            for dep in &spec.deps {
                store.record(Event::EdgeAdded {
                    plan_id: plan_id.clone(),
                    from_node: dep.clone(),
                    to_node: spec.node_id.clone(),
                    ts: at(),
                })?;
            }
        }
        plan_ids.push(plan_id);
    }

    store.funnel.promote_ready();
    Ok(Some((idea_id, plan_ids)))
}

/// Test-friendly thin wrapper: inject a `size`-plan demo (one plan per "size"
/// is overkill; instead inject a single plan whose node count scales with
/// `size`) into an already-open store. Used by `tests/funnel_demo.rs`.
///
/// Layout chosen so the golden set is easy to reason about:
///   * 1 plan,
///   * `size * 3` nodes,
///   * 3 layers,
///   * fixed seed `0xF1` (so the expected DAG is stable across runs).
pub fn inject_demo(store: &mut Store, size: usize) -> Result<Option<(IdeaId, Vec<PlanId>)>> {
    inject_dep_graph(store, 1, size.max(1) * 3, 3, 0xF1)
}

// ─── on-disk fake-repo materialisation ──────────────────────────────────────

/// Write one fake repo dir: `Cargo.toml` (path-deps on lower-indexed siblings),
/// `src/lib.rs`, and a real git repo (`git init` + add + commit). Idempotent:
/// existing files are overwritten, an existing `.git` is left as-is, and the
/// commit step is best-effort (a clean tree re-commit is a harmless no-op).
fn write_fake_repo(dir: &Path, size: usize, i: usize) -> Result<()> {
    let src = dir.join("src");
    std::fs::create_dir_all(&src)
        .with_context(|| format!("create {}", src.display()))?;

    // Path-deps on every LOWER-indexed sibling → a layered (acyclic) dep graph
    // that mirrors the funnel DAG. Sibling location depends on the layout.
    let name = crate_name(i);
    let mut deps = String::new();
    for j in 1..i {
        let sibling = crate_name(j);
        // For size==1 there are no siblings (i==1 only); for size>1 siblings are
        // adjacent dirs under the same container.
        let rel = format!("../{sibling}");
        deps.push_str(&format!("{sibling} = {{ path = \"{rel}\" }}\n"));
    }

    let cargo = format!(
        "[package]\n\
         name = \"{name}\"\n\
         version = \"0.1.0\"\n\
         edition = \"2021\"\n\
         publish = false\n\
         \n\
         # Generated by `nornir funnel demo` (size {size}). Safe to delete via\n\
         # `nornir funnel nuke --yes`.\n\
         [dependencies]\n\
         {deps}"
    );
    std::fs::write(dir.join("Cargo.toml"), cargo)
        .with_context(|| format!("write Cargo.toml in {}", dir.display()))?;

    let lib = format!(
        "//! Fake demo crate `{name}` — generated by `nornir funnel demo`.\n\
         //! Part of a layered path-dep graph mirroring the funnel demo DAG.\n\
         \n\
         /// Returns this crate's 1-based index in the demo dep graph.\n\
         pub fn index() -> usize {{\n    {i}\n}}\n"
    );
    std::fs::write(src.join("lib.rs"), lib)
        .with_context(|| format!("write src/lib.rs in {}", dir.display()))?;

    git_init_commit(dir)?;
    Ok(())
}

/// `git init` + add-all + commit, idempotent and best-effort. Uses the system
/// `git` binary via `std::process::Command`; failures (e.g. no git, or nothing
/// to commit) are downgraded to warnings so the demo still produces files.
fn git_init_commit(dir: &Path) -> Result<()> {
    use std::process::Command;

    let run = |args: &[&str]| -> std::io::Result<std::process::Output> {
        Command::new("git")
            .args(args)
            .current_dir(dir)
            // Force a deterministic identity + no global config interference so
            // the commit succeeds in CI/sandbox where user.name may be unset.
            .env("GIT_AUTHOR_NAME", "nornir-funnel-demo")
            .env("GIT_AUTHOR_EMAIL", "demo@nornir.local")
            .env("GIT_COMMITTER_NAME", "nornir-funnel-demo")
            .env("GIT_COMMITTER_EMAIL", "demo@nornir.local")
            .output()
    };

    if !dir.join(".git").exists() {
        let _ = run(&["init", "-q"]);
        let _ = run(&["checkout", "-q", "-B", "main"]);
    }
    let _ = run(&["add", "-A"]);
    // Commit only if there's something staged (re-running on a clean tree is a
    // no-op we silently swallow).
    let _ = run(&[
        "-c",
        "commit.gpgsign=false",
        "commit",
        "-q",
        "-m",
        "funnel demo snapshot",
    ]);
    Ok(())
}

/// Materialise `--size N` fake repos under the git root AND inject the matching
/// funnel DAG into the warehouse at `funnel_root`. Writes the manifest. Both
/// halves are idempotent.
///
/// Returns the [`DemoManifest`] describing what was created on disk.
pub fn run_demo(funnel_root: &Path, size: usize) -> Result<DemoManifest> {
    let size = size.max(1);
    let git_root = resolve_git_root(funnel_root);
    std::fs::create_dir_all(&git_root)
        .with_context(|| format!("create git root {}", git_root.display()))?;

    // --- disk half: fake repos + manifest ---
    let mut repos = Vec::with_capacity(size);
    for i in 1..=size {
        let dir = repo_dir(&git_root, size, i);
        write_fake_repo(&dir, size, i)?;
        repos.push(dir);
    }
    let container = if size == 1 {
        Some(container_dir(&git_root, size))
    } else {
        Some(container_dir(&git_root, size))
    };
    let manifest = DemoManifest {
        kind: "nornir-funnel-demo".to_string(),
        created_at: Utc::now(),
        size,
        repos,
        container,
    };
    manifest.save(&git_root)?;

    // --- warehouse half: the mirroring funnel DAG ---
    // One plan whose nodes are the fake repos, edges = path-deps. Layers derived
    // from index so node i depends only on lower-indexed nodes (acyclic).
    let mut store = Store::open(funnel_root)
        .with_context(|| format!("open funnel store at {}", funnel_root.display()))?;
    inject_repo_dag(&mut store, size)?;

    Ok(manifest)
}

/// Inject a funnel DAG that mirrors the on-disk repo dep graph: one node per
/// fake repo `i`, one edge `j -> i` for each path-dep (`j < i`). Idempotent on
/// the `funnel:demo` source tag.
fn inject_repo_dag(store: &mut Store, size: usize) -> Result<Option<(IdeaId, PlanId)>> {
    if store.funnel.ideas.values().any(|i| i.source == DEMO_SOURCE) {
        return Ok(None);
    }
    let base = Utc::now();
    let mut tick = 0i64;
    let mut at = || {
        tick += 1;
        base + chrono::Duration::microseconds(tick)
    };

    let idea_id = IdeaId::seq(store.funnel.next_idea.max(1));
    store.record(Event::IdeaSubmitted {
        id: idea_id.clone(),
        source: DEMO_SOURCE.to_string(),
        text: format!("funnel demo: {size} fake repo(s) with a layered path-dep graph"),
        refs: vec![],
        ts: at(),
    })?;
    let plan_id = PlanId::seq(store.funnel.next_plan.max(1));
    store.record(Event::PlanCreated {
        id: plan_id.clone(),
        idea_id: idea_id.clone(),
        summary: format!("demo: {size} fake repos (path-dep DAG)"),
        planner: DEMO_SOURCE.to_string(),
        ts: at(),
    })?;
    store.record(Event::PlanStatusChanged {
        plan_id: plan_id.clone(),
        status: PlanStatus::Active,
        why: None,
        ts: at(),
    })?;

    // node[i] for fake repo i (1-based). Record nodes one at a time, capturing
    // the monotonic seq id each `NodeAdded` consumed (ids must stay stable so
    // the edges below reference the right nodes).
    let mut ids = Vec::with_capacity(size);
    for i in 1..=size {
        let nid = NodeId::seq(store.funnel.next_node.max(1));
        store.record(Event::NodeAdded {
            plan_id: plan_id.clone(),
            node_id: nid.clone(),
            kind: "code:write".to_string(),
            params: serde_json::Map::new(),
            targets: vec![format!("{}/src/lib.rs", crate_name(i))],
            prompt_excerpt: None,
            ts: at(),
        })?;
        ids.push(nid);
    }
    // edges: repo i path-deps on every j<i → edge node[j] -> node[i].
    for i in 1..=size {
        for j in 1..i {
            store.record(Event::EdgeAdded {
                plan_id: plan_id.clone(),
                from_node: ids[j - 1].clone(),
                to_node: ids[i - 1].clone(),
                ts: at(),
            })?;
        }
    }
    store.funnel.promote_ready();
    Ok(Some((idea_id, plan_id)))
}

// ─── nuke (disk-only) ───────────────────────────────────────────────────────

/// What a nuke *would* do — the manifest's repos + container — without deleting.
/// Returns `None` when no manifest is present (nothing to nuke).
pub fn nuke_plan(funnel_root: &Path) -> Option<DemoManifest> {
    let git_root = resolve_git_root(funnel_root);
    DemoManifest::load(&git_root)
}

/// Delete ONLY the paths the manifest recorded (the fake repos + their
/// container + the manifest file). **Disk-only**: the warehouse funnel data is
/// never touched. Returns the list of removed dirs.
///
/// Safe by construction: it deletes exactly the manifest's `repos`/`container`,
/// not a glob — a manually-created dir that happens to match the naming scheme
/// is left alone unless the manifest lists it.
pub fn nuke_disk(funnel_root: &Path) -> Result<Vec<PathBuf>> {
    let git_root = resolve_git_root(funnel_root);
    let Some(manifest) = DemoManifest::load(&git_root) else {
        return Ok(Vec::new());
    };
    let mut removed = Vec::new();
    for repo in &manifest.repos {
        if repo.exists() {
            std::fs::remove_dir_all(repo)
                .with_context(|| format!("remove fake repo {}", repo.display()))?;
            removed.push(repo.clone());
        }
    }
    // Remove the container dir if it is now empty (only our repos lived there).
    if let Some(container) = &manifest.container {
        if container.exists() {
            let empty = std::fs::read_dir(container)
                .map(|mut d| d.next().is_none())
                .unwrap_or(false);
            if empty {
                let _ = std::fs::remove_dir(container);
            }
        }
    }
    // Finally drop the manifest itself.
    let _ = std::fs::remove_file(DemoManifest::path(&git_root));
    Ok(removed)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The dep-graph spec is acyclic: every edge goes lower-layer → higher-layer.
    #[test]
    fn spec_edges_are_layer_monotone() {
        let specs = build_dep_graph_spec(2, 12, 4, 0xABCD);
        for plan in &specs {
            let layer_of: std::collections::HashMap<&NodeId, usize> =
                plan.iter().map(|s| (&s.node_id, s.layer)).collect();
            for s in plan {
                for dep in &s.deps {
                    let dl = layer_of[dep];
                    assert!(dl < s.layer, "edge {dep} -> {} not layer-monotone", s.node_id);
                }
            }
        }
    }

    /// Same seed → identical spec (determinism), different seed → may differ.
    #[test]
    fn spec_is_deterministic() {
        let a = build_dep_graph_spec(1, 9, 3, 7);
        let b = build_dep_graph_spec(1, 9, 3, 7);
        assert_eq!(a.len(), b.len());
        for (pa, pb) in a.iter().zip(&b) {
            assert_eq!(pa.len(), pb.len());
            for (na, nb) in pa.iter().zip(pb) {
                assert_eq!(na.node_id, nb.node_id);
                assert_eq!(na.deps, nb.deps);
                assert_eq!(na.layer, nb.layer);
            }
        }
    }

    /// At least one root (layer 0, no deps) always exists, so `promote_ready`
    /// has somewhere to start.
    #[test]
    fn spec_has_a_root() {
        let specs = build_dep_graph_spec(1, 20, 5, 1);
        assert!(specs[0].iter().any(|s| s.layer == 0 && s.deps.is_empty()));
    }
}