nornir 0.5.0

//! Pure-Rust symbol + call + feature-gate extractor.
//!
//! For every `Cargo.toml` under the repo we walk every `.rs` under
//! src/ tests/ benches/ examples/, parse with `syn`, and emit:
//!
//!  - `SymbolRow`      — one per item (fn / struct / enum / trait / impl-fn / mod / type / const / static).
//!  - `CallEdgeRow`    — one per call expression inside an fn body.
//!  - `FeatureGateRow` — one per `#[cfg(...)]` attribute on a discovered item.
//!
//! Limitations vs rustdoc-json:
//!   - No type resolution — `callee_ident` is the raw name.
//!   - No re-exports / generic impl expansion.
//!   - `line` is the 1-based source line of the item name / call site
//!     (via proc-macro2 `span-locations`).

use std::path::Path;

use anyhow::Result;
use chrono::{DateTime, Utc};
use syn::visit::Visit;
use syn::{Attribute, Expr, ImplItem, Item, ItemImpl, Meta, Visibility};
use syn::spanned::Spanned;
use uuid::Uuid;
use walkdir::WalkDir;
use znippy_zoomies::gatling_forkjoin::gatling_map_balanced;

#[derive(Debug, Default)]
pub struct SymbolScan {
    pub snapshot_id: Uuid,
    pub ts: DateTime<Utc>,
    pub repo: String,
    pub symbols: Vec<SymbolRow>,
    pub calls: Vec<CallEdgeRow>,
    pub features: Vec<FeatureGateRow>,
    /// Every `#[test]`/`#[tokio::test]`/`#[test_case]`/`#[rstest]` fn discovered
    /// during the (BUILD-FREE) syn parse — the test INVENTORY. This is what lets
    /// the 🧪 matrix show every test as `X` (not-run) the instant a workspace is
    /// added, with zero `cargo build` (discovery is free; only EXECUTION needs a
    /// build, and that stays gated to `nornir release` / `nornir test run`).
    pub tests: Vec<TestDefRow>,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct SymbolRow {
    pub crate_name: String,
    pub module_path: String,
    pub item_kind: String,
    pub item_name: String,
    pub visibility: String,
    pub file: String,
    pub line: u32,
    pub doc_lines: u32,
    pub signature: Option<String>,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct CallEdgeRow {
    pub crate_name: String,
    pub caller_path: String,
    pub callee_ident: String,
    pub call_kind: String,
    pub file: String,
    pub line: u32,
}

/// One discovered test function — the build-free test INVENTORY row. Emitted by
/// the syn walker for any `fn` carrying a test attribute (`#[test]`,
/// `#[tokio::test]`, `#[test_case(...)]`, `#[rstest]`, …). Heavy tests (the
/// `#[ignore]`-marked container/corpus/matrix suites) are flagged `is_heavy` so
/// the viz can tag them and the gating audit can keep them out of the poll loop.
#[derive(Debug, Clone, serde::Serialize)]
pub struct TestDefRow {
    pub crate_name: String,
    pub module_path: String,
    pub test_name: String,
    pub file: String,
    pub line: u32,
    /// `#[ignore]`-marked → the heavy container/corpus/matrix tests. These run
    /// ONLY in `nornir release` (default) or on explicit `nornir test run`, never
    /// on initial load or in the poll loop.
    pub is_heavy: bool,
    /// `async fn` and/or `#[tokio::test]` — informational, surfaced in the viz.
    pub is_async: bool,
}

#[derive(Debug, Clone)]
pub struct FeatureGateRow {
    pub crate_name: String,
    pub module_path: String,
    pub item_name: String,
    pub cfg_expr: String,
    pub file: String,
    pub line: u32,
}

pub fn scan_repo(
    repo_root: &Path,
    repo_name: &str,
    snapshot_id: Uuid,
    ts: DateTime<Utc>,
) -> Result<SymbolScan> {
    // 1. Discover every (file, crate, section, size) — a fast SERIAL walk (I/O only).
    let mut files: Vec<(std::path::PathBuf, String, &'static str, u64)> = Vec::new();
    discover_repo_files(repo_root, &mut files);

    // 2. Parse every file across ALL cores — the "(TODO) gatling over FILES for the
    //    member scan" finished. `scan_repo` is always called TOP-LEVEL (deep_scan
    //    parses via the file-granular `scan_file` through its OWN gatling pool,
    //    never this fn; the monitor's republish uses the flattened
    //    `scan_repos_flattened` instead of looping this), so an inner pool here
    //    never nests/oversubscribes. `syn::parse_file` (in `scan_one_file`) is the cost.
    Ok(parse_files_parallel(repo_root, repo_name, snapshot_id, ts, &files))
}

/// The serial I/O discovery half of a repo scan: append every scannable `.rs`
/// file of every crate under `repo_root` as `(file, crate, section, size)`. Pure
/// directory walk (no parsing), shared by the per-repo [`scan_repo`] and the
/// flattened multi-repo [`scan_repos_flattened`]. A vanished/again-unreadable dir
/// entry is skipped (best-effort), never fatal.
fn discover_repo_files(
    repo_root: &Path,
    files: &mut Vec<(std::path::PathBuf, String, &'static str, u64)>,
) {
    for entry in WalkDir::new(repo_root)
        .into_iter()
        .filter_entry(|e| !is_skipped_dir(&e.file_name().to_string_lossy()))
    {
        let Ok(entry) = entry else { continue };
        if entry.file_name() == "Cargo.toml" {
            collect_crate_files(entry.path(), files);
        }
    }
}

/// **Flattened multi-repo symbol scan** — the monitor republish step-1 fan.
///
/// Runs ONE balanced gatling pool over the UNION of every repo's source files
/// (heaviest-first LPT across the *whole* set, not per repo), then groups the
/// per-file parses back into one [`SymbolScan`] per repo. So 1 huge member or 10
/// small ones both saturate every core, and size skew is auto-balanced ACROSS
/// members — and there is exactly ONE pool, never a per-member pool nested under
/// a member loop (the oversubscription this avoids).
///
/// `repos` is `(repo_name, repo_root)`; `snapshot_ids[i]` keys repo `i`'s scan so
/// each member's warehouse rows/ledger stay per-member. The work list holds only
/// lightweight `(idx, path, crate, section, size)` tags — file bytes are read
/// INSIDE each worker (`scan_one_file`), never copied up front. Returns one scan
/// per repo, in the input order of `repos`. Top-level only (it IS the pool).
pub fn scan_repos_flattened(
    repos: &[(String, std::path::PathBuf)],
    snapshot_ids: &[Uuid],
    ts: DateTime<Utc>,
) -> Vec<SymbolScan> {
    assert_eq!(repos.len(), snapshot_ids.len(), "one snapshot id per repo");

    // 1. Serial I/O discovery → union of (repo_idx, file, crate, section, size).
    //    Tags only; no file contents are read here.
    let mut union: Vec<(usize, std::path::PathBuf, String, &'static str, u64)> = Vec::new();
    for (i, (_, root)) in repos.iter().enumerate() {
        let mut files: Vec<(std::path::PathBuf, String, &'static str, u64)> = Vec::new();
        discover_repo_files(root, &mut files);
        for (file, crate_name, section, size) in files {
            union.push((i, file, crate_name, section, size));
        }
    }

    // 2. ONE balanced pool over the whole union (LPT by byte size). Each worker
    //    reads + parses its claimed file and tags the result with the repo index.
    let parsed = gatling_map_balanced(
        &union,
        scan_worker_budget(),
        SCAN_BATCH,
        |t| t.4,
        |_, (idx, file, crate_name, section, _)| {
            let mut one = SymbolScan {
                snapshot_id: snapshot_ids[*idx],
                ts,
                repo: repos[*idx].0.clone(),
                ..Default::default()
            };
            scan_one_file(&repos[*idx].1, file, crate_name, section, &mut one);
            (*idx, one)
        },
    );

    // 3. Group per-file scans back per repo (parsed is in original union order ⇒
    //    deterministic, identical to scanning each member serially).
    let mut out: Vec<SymbolScan> = repos
        .iter()
        .enumerate()
        .map(|(i, (name, _))| SymbolScan {
            snapshot_id: snapshot_ids[i],
            ts,
            repo: name.clone(),
            ..Default::default()
        })
        .collect();
    for (idx, mut one) in parsed {
        out[idx].symbols.append(&mut one.symbols);
        out[idx].calls.append(&mut one.calls);
        out[idx].features.append(&mut one.features);
        out[idx].tests.append(&mut one.tests);
    }
    out
}

/// Batch size (files claimed per atomic steal) handed to the gatling pool — large
/// enough to amortize the counter hand-off across the small-file tail, small enough
/// to keep the heaviest-first schedule balanced. Matches the zoomies guide's worked
/// example for a per-file `syn` parse.
const SCAN_BATCH: usize = 8;

/// Worker budget for the member scan: `NORNIR_SCAN_THREADS` if set (≥1) — an escape
/// hatch for A/B perf measurement and for capping oversubscription — else one worker
/// per core. Mirrors the `LBZIP2_THREADS` convention in the gatling family. Passed as
/// `n_workers` to [`gatling_map_balanced`] (whose own `0` ⇒ all-cores default we
/// bypass so this env override always wins).
fn scan_worker_budget() -> usize {
    std::env::var("NORNIR_SCAN_THREADS")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|&n| n >= 1)
        .unwrap_or_else(|| {
            std::thread::available_parallelism()
                .map(|w| w.get())
                .unwrap_or(1)
        })
}

/// Parse the already-discovered `files` into one merged [`SymbolScan`], spreading
/// the per-file `syn::parse_file` CPU cost across all cores through the canonical
/// znippy-zoomies pool [`gatling_map_balanced`] (LPT by byte size, heaviest-first,
/// batched self-dispatch, owned per-file scans back in original order). The pool was
/// de-inlined into zoomies 0.1.13; the worked rationale lives in the zoomies gatling
/// docs. Call TOP-LEVEL only — it IS the pool; nesting it under another pool
/// oversubscribes.
fn parse_files_parallel(
    repo_root: &Path,
    repo_name: &str,
    snapshot_id: Uuid,
    ts: DateTime<Utc>,
    files: &[(std::path::PathBuf, String, &'static str, u64)],
) -> SymbolScan {
    let mut out = SymbolScan {
        snapshot_id,
        ts,
        repo: repo_name.to_string(),
        ..Default::default()
    };
    // ONE balanced pool over this repo's files (LPT by byte size); merge per-file
    // scans back in discovery order ⇒ deterministic, identical to serial.
    let parsed = gatling_map_balanced(files, scan_worker_budget(), SCAN_BATCH, |f| f.3, |_, (file, crate_name, section, _)| {
        let mut one = SymbolScan {
            snapshot_id,
            ts,
            repo: repo_name.to_string(),
            ..Default::default()
        };
        scan_one_file(repo_root, file, crate_name, section, &mut one);
        one
    });
    for mut one in parsed {
        out.symbols.append(&mut one.symbols);
        out.calls.append(&mut one.calls);
        out.features.append(&mut one.features);
        out.tests.append(&mut one.tests);
    }
    out
}

fn is_skipped_dir(name: &str) -> bool {
    matches!(name, "target" | ".git" | "node_modules" | ".nornir")
}

/// Collect every scannable `.rs` file of the crate at `cargo_toml` as
/// `(file, crate_name, section)` tuples — the serial discovery half, so the
/// `syn` parse half ([`scan_repo`] step 2) can run across all cores.
fn collect_crate_files(
    cargo_toml: &Path,
    files: &mut Vec<(std::path::PathBuf, String, &'static str, u64)>,
) {
    let Ok(text) = std::fs::read_to_string(cargo_toml) else { return };
    let Ok(doc) = text.parse::<toml::Value>() else { return };
    let Some(crate_name) = doc
        .get("package")
        .and_then(|p| p.get("name"))
        .and_then(|n| n.as_str())
    else { return };
    let Some(crate_dir) = cargo_toml.parent() else { return };

    for sub in &["src", "tests", "benches", "examples"] {
        let dir = crate_dir.join(sub);
        if !dir.is_dir() { continue; }
        for entry in WalkDir::new(&dir)
            .into_iter()
            .filter_entry(|e| !is_skipped_dir(&e.file_name().to_string_lossy()))
        {
            let Ok(entry) = entry else { continue };
            if entry.file_type().is_file()
                && entry.path().extension().and_then(|e| e.to_str()) == Some("rs")
            {
                // Stat the size now (the discovery walk is already touching the
                // dir entry) so the parse pool can schedule heaviest-file-first.
                let size = entry.metadata().map(|m| m.len()).unwrap_or(0);
                files.push((entry.path().to_path_buf(), crate_name.to_string(), *sub, size));
            }
        }
    }
}

fn scan_one_file(
    repo_root: &Path,
    file: &Path,
    crate_name: &str,
    section: &str,
    out: &mut SymbolScan,
) {
    let Ok(text) = std::fs::read_to_string(file) else { return };
    let rel = file.strip_prefix(repo_root).unwrap_or(file).to_string_lossy().to_string();
    let stem = file.file_stem().and_then(|s| s.to_str()).unwrap_or("");
    scan_content(&text, &rel, crate_name, section, stem, out);
}

/// Parse one already-read `.rs` source (the disk-free core of [`scan_one_file`])
/// — used both by the on-disk walk and the in-memory `.crate` tarball path
/// ([`scan_crate_entries`]). `rel` is the file's path for the symbol rows; the
/// `(section, stem)` derive the base module.
fn scan_content(
    text: &str,
    rel: &str,
    crate_name: &str,
    section: &str,
    stem: &str,
    out: &mut SymbolScan,
) {
    let Ok(syntax) = syn::parse_file(text) else { return };
    let base_module = if section == "src" && (stem == "lib" || stem == "main") {
        crate_name.to_string()
    } else if section == "src" {
        format!("{crate_name}::{stem}")
    } else {
        format!("{crate_name}::{section}::{stem}")
    };
    let mut v = Walker {
        crate_name: crate_name.to_string(),
        file: rel.to_string(),
        module_stack: vec![base_module],
        out,
    };
    v.walk_items(&syntax.items);
}

/// Parse ONE crate file (already decompressed, in memory) → its symbols/calls/
/// features. This is the **file-granular** unit for deep_scan's parse stage, so a
/// giant crate's files spread across all cores instead of pinning one worker.
/// `tar_path` is like `serde-1.0.219/src/lib.rs`; non-`.rs` or files outside
/// `src/tests/benches/examples` yield an empty scan. Rows self-tag with
/// `crate_name`, so the caller needs NO per-crate reassembly.
pub fn scan_file(
    crate_name: &str,
    tar_path: &str,
    content: &str,
    repo_name: &str,
    snapshot_id: Uuid,
    ts: DateTime<Utc>,
) -> SymbolScan {
    let mut out = SymbolScan {
        snapshot_id,
        ts,
        repo: repo_name.to_string(),
        ..Default::default()
    };
    if !tar_path.ends_with(".rs") {
        return out;
    }
    // Strip the leading `<name-version>/` component → `src/lib.rs`.
    let rel = tar_path.splitn(2, '/').nth(1).unwrap_or(tar_path);
    let section = rel.split('/').next().unwrap_or("");
    if !matches!(section, "src" | "tests" | "benches" | "examples") {
        return out;
    }
    let stem = Path::new(rel).file_stem().and_then(|s| s.to_str()).unwrap_or("");
    scan_content(content, rel, crate_name, section, stem, &mut out);
    out
}

/// Scan a crate straight from its **in-memory `.crate` tar entries** (gzip
/// already decompressed by `lgz`), with NO disk reads. Each `entries` item is
/// `(tar_path, bytes)` where `tar_path` looks like `serde-1.0.219/src/lib.rs`;
/// only `.rs` under `src/tests/benches/examples` is parsed. This is the
/// gatling-worker path for `deep_scan`: one fat CPU unit per crate
/// (decompress + parse), so 32 workers each chew a whole crate.
pub fn scan_crate_entries(
    crate_name: &str,
    entries: &[(String, Vec<u8>)],
    repo_name: &str,
    snapshot_id: Uuid,
    ts: DateTime<Utc>,
) -> SymbolScan {
    let mut out = SymbolScan {
        snapshot_id,
        ts,
        repo: repo_name.to_string(),
        ..Default::default()
    };
    for (path, bytes) in entries {
        if !path.ends_with(".rs") {
            continue;
        }
        // Strip the leading `<name-version>/` component → `src/lib.rs`.
        let rel = path.splitn(2, '/').nth(1).unwrap_or(path);
        let section = rel.split('/').next().unwrap_or("");
        if !matches!(section, "src" | "tests" | "benches" | "examples") {
            continue;
        }
        let Ok(text) = std::str::from_utf8(bytes) else { continue };
        let stem = Path::new(rel).file_stem().and_then(|s| s.to_str()).unwrap_or("");
        scan_content(text, rel, crate_name, section, stem, &mut out);
    }
    out
}

struct Walker<'a> {
    crate_name: String,
    file: String,
    module_stack: Vec<String>,
    out: &'a mut SymbolScan,
}

impl<'a> Walker<'a> {
    fn cur(&self) -> &str {
        self.module_stack.last().map(|s| s.as_str()).unwrap_or("")
    }

    fn walk_items(&mut self, items: &[Item]) {
        for item in items {
            self.walk_item(item);
        }
    }

    fn walk_item(&mut self, item: &Item) {
        match item {
            Item::Fn(f) => {
                self.emit("fn", &f.sig.ident.to_string(), &vis(&f.vis),
                    Some(format!("fn {}{}", f.sig.ident, fn_inputs(&f.sig.inputs))),
                    &f.attrs, f.sig.ident.span());
                self.maybe_emit_test(&f.sig.ident.to_string(), &f.attrs, f.sig.asyncness.is_some(), f.sig.ident.span());
                self.walk_fn_body(&f.sig.ident.to_string(), &f.block);
            }
            Item::Struct(s) => self.emit("struct", &s.ident.to_string(), &vis(&s.vis), None, &s.attrs, s.ident.span()),
            Item::Enum(e)   => self.emit("enum",   &e.ident.to_string(), &vis(&e.vis), None, &e.attrs, e.ident.span()),
            Item::Trait(t)  => self.emit("trait",  &t.ident.to_string(), &vis(&t.vis), None, &t.attrs, t.ident.span()),
            Item::Type(t)   => self.emit("type",   &t.ident.to_string(), &vis(&t.vis), None, &t.attrs, t.ident.span()),
            Item::Const(c)  => self.emit("const",  &c.ident.to_string(), &vis(&c.vis), None, &c.attrs, c.ident.span()),
            Item::Static(s) => self.emit("static", &s.ident.to_string(), &vis(&s.vis), None, &s.attrs, s.ident.span()),
            Item::Macro(m) => {
                if let Some(id) = &m.ident {
                    self.emit("macro", &id.to_string(), "pub", None, &m.attrs, id.span());
                }
            }
            Item::Mod(m) => {
                let nested = format!("{}::{}", self.cur(), m.ident);
                self.emit("mod", &m.ident.to_string(), &vis(&m.vis), None, &m.attrs, m.ident.span());
                if let Some((_, sub)) = &m.content {
                    self.module_stack.push(nested);
                    self.walk_items(sub);
                    self.module_stack.pop();
                }
            }
            Item::Impl(i) => self.walk_impl(i),
            _ => {}
        }
    }

    fn walk_impl(&mut self, i: &ItemImpl) {
        let self_ty = tokens_to_string(&i.self_ty);
        let trait_part = i.trait_.as_ref()
            .map(|(_, p, _)| path_to_string(p))
            .unwrap_or_default();
        let impl_label = if trait_part.is_empty() {
            format!("impl {self_ty}")
        } else {
            format!("impl {trait_part} for {self_ty}")
        };
        self.emit("impl", &impl_label, "pub", None, &i.attrs, i.self_ty.span());
        let parent = self.cur().to_string();
        self.module_stack.push(format!("{parent}::{self_ty}"));
        for it in &i.items {
            if let ImplItem::Fn(f) = it {
                self.emit("fn", &f.sig.ident.to_string(), &vis(&f.vis),
                    Some(format!("fn {}{}", f.sig.ident, fn_inputs(&f.sig.inputs))),
                    &f.attrs, f.sig.ident.span());
                self.maybe_emit_test(&f.sig.ident.to_string(), &f.attrs, f.sig.asyncness.is_some(), f.sig.ident.span());
                self.walk_fn_body(&f.sig.ident.to_string(), &f.block);
            }
        }
        self.module_stack.pop();
    }

    fn walk_fn_body(&mut self, fn_name: &str, body: &syn::Block) {
        let caller = format!("{}::{}", self.cur(), fn_name);
        let mut cv = CallCollector {
            caller,
            crate_name: self.crate_name.clone(),
            file: self.file.clone(),
            out: &mut self.out.calls,
        };
        cv.visit_block(body);
    }

    fn emit(
        &mut self,
        kind: &str,
        name: &str,
        visi: &str,
        sig: Option<String>,
        attrs: &[Attribute],
        span: proc_macro2::Span,
    ) {
        let line = span.start().line as u32;
        let doc_lines = attrs.iter().filter(|a| a.path().is_ident("doc")).count() as u32;
        self.out.symbols.push(SymbolRow {
            crate_name: self.crate_name.clone(),
            module_path: self.cur().to_string(),
            item_kind: kind.to_string(),
            item_name: name.to_string(),
            visibility: visi.to_string(),
            file: self.file.clone(),
            line,
            doc_lines,
            signature: sig,
        });
        for cfg in extract_cfgs(attrs) {
            self.out.features.push(FeatureGateRow {
                crate_name: self.crate_name.clone(),
                module_path: self.cur().to_string(),
                item_name: name.to_string(),
                cfg_expr: cfg,
                file: self.file.clone(),
                line,
            });
        }
    }

    /// If `attrs` mark this fn as a test (`#[test]`, `#[tokio::test]`,
    /// `#[test_case(...)]`, `#[rstest]`, `#[async_std::test]`, …), push one
    /// [`TestDefRow`] into the inventory. `#[ignore]` → `is_heavy`. This is the
    /// build-free discovery: it runs in the SAME syn parse as symbols/calls, so a
    /// fresh workspace's whole test surface lands without compiling anything.
    fn maybe_emit_test(&mut self, name: &str, attrs: &[Attribute], asyncness: bool, span: proc_macro2::Span) {
        if !is_test_attr(attrs) {
            return;
        }
        let is_heavy = attrs.iter().any(|a| a.path().is_ident("ignore"));
        let is_async = asyncness || attrs.iter().any(|a| attr_last_seg(a) == "tokio" || attr_is_async_test(a));
        self.out.tests.push(TestDefRow {
            crate_name: self.crate_name.clone(),
            module_path: self.cur().to_string(),
            test_name: name.to_string(),
            file: self.file.clone(),
            line: span.start().line as u32,
            is_heavy,
            is_async,
        });
    }
}

/// Does any attribute mark this fn as a test? Matches the common runners:
/// `#[test]`, `#[tokio::test]`, `#[async_std::test]`, `#[test_case(...)]`,
/// `#[rstest]`, `#[rstest::rstest]`, `#[googletest::test]`, `#[proptest]`. We
/// match on the path's LAST segment being `test`/`rstest`/`test_case`/`proptest`
/// so namespaced forms (`tokio::test`) are caught without resolving the path.
fn is_test_attr(attrs: &[Attribute]) -> bool {
    attrs.iter().any(|a| {
        let last = attr_last_seg(a);
        matches!(last.as_str(), "test" | "rstest" | "test_case" | "proptest")
    })
}

/// `#[tokio::test]` / `#[async_std::test]` — an async-flavoured test attr.
fn attr_is_async_test(a: &Attribute) -> bool {
    let segs: Vec<String> = a.path().segments.iter().map(|s| s.ident.to_string()).collect();
    matches!(segs.as_slice(), [first, second] if (first == "tokio" || first == "async_std") && second == "test")
}

/// The last `::`-segment of an attribute path (`tokio::test` → `test`).
fn attr_last_seg(a: &Attribute) -> String {
    a.path().segments.last().map(|s| s.ident.to_string()).unwrap_or_default()
}

struct CallCollector<'a> {
    caller: String,
    crate_name: String,
    file: String,
    out: &'a mut Vec<CallEdgeRow>,
}

impl<'ast, 'a> Visit<'ast> for CallCollector<'a> {
    fn visit_expr(&mut self, e: &'ast Expr) {
        match e {
            Expr::Call(c) => {
                let ident = match &*c.func {
                    Expr::Path(p) => path_to_string(&p.path),
                    other => tokens_to_string(other),
                };
                self.out.push(CallEdgeRow {
                    crate_name: self.crate_name.clone(),
                    caller_path: self.caller.clone(),
                    callee_ident: ident,
                    call_kind: "call".into(),
                    file: self.file.clone(),
                    line: c.func.span().start().line as u32,
                });
            }
            Expr::MethodCall(m) => {
                self.out.push(CallEdgeRow {
                    crate_name: self.crate_name.clone(),
                    caller_path: self.caller.clone(),
                    callee_ident: m.method.to_string(),
                    call_kind: "method".into(),
                    file: self.file.clone(),
                    line: m.method.span().start().line as u32,
                });
            }
            _ => {}
        }
        syn::visit::visit_expr(self, e);
    }
}

fn vis(v: &Visibility) -> String {
    match v {
        Visibility::Public(_) => "pub".into(),
        Visibility::Restricted(r) => format!("pub({})", path_to_string(&r.path)),
        Visibility::Inherited => "priv".into(),
    }
}

fn path_to_string(p: &syn::Path) -> String {
    p.segments.iter().map(|s| s.ident.to_string()).collect::<Vec<_>>().join("::")
}

fn tokens_to_string<T: quote::ToTokens>(t: &T) -> String {
    let mut ts = proc_macro2::TokenStream::new();
    t.to_tokens(&mut ts);
    ts.to_string()
}

fn fn_inputs(inputs: &syn::punctuated::Punctuated<syn::FnArg, syn::Token![,]>) -> String {
    let parts: Vec<String> = inputs.iter().map(|a| match a {
        syn::FnArg::Receiver(_) => "self".into(),
        syn::FnArg::Typed(t) => tokens_to_string(&t.ty),
    }).collect();
    format!("({})", parts.join(", "))
}

fn extract_cfgs(attrs: &[Attribute]) -> Vec<String> {
    let mut out = Vec::new();
    for a in attrs {
        if a.path().is_ident("cfg") {
            if let Meta::List(l) = &a.meta {
                out.push(l.tokens.to_string());
            }
        }
    }
    out
}

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

    #[test]
    fn scans_basic_crate() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::create_dir_all(root.join("src")).unwrap();
        std::fs::write(root.join("Cargo.toml"), r#"[package]
name = "demo"
version = "0.1.0"
edition = "2021"
"#).unwrap();
        std::fs::write(root.join("src/lib.rs"), r#"
pub fn outer(x: u32) -> u32 {
    helper(x) + 1
}
fn helper(x: u32) -> u32 { x.saturating_add(2) }

pub struct Foo { pub n: u32 }

impl Foo {
    pub fn make() -> Self { Self { n: 0 } }
}

pub trait Bar { fn baz(&self); }

#[cfg(feature = "experimental")]
pub fn gated() {}
"#).unwrap();

        let scan = scan_repo(root, "demo", Uuid::new_v4(), Utc::now()).unwrap();
        // Symbols: outer, helper, Foo, impl Foo, make, Bar, gated, baz? (trait fn is not Item)
        let kinds: Vec<_> = scan.symbols.iter().map(|s| (s.item_kind.as_str(), s.item_name.as_str())).collect();
        assert!(kinds.contains(&("fn", "outer")), "{kinds:?}");
        assert!(kinds.contains(&("fn", "helper")));
        assert!(kinds.contains(&("struct", "Foo")));
        assert!(kinds.iter().any(|(k, _)| *k == "impl"));
        assert!(kinds.contains(&("fn", "make")));
        assert!(kinds.contains(&("trait", "Bar")));
        assert!(kinds.contains(&("fn", "gated")));

        // Calls: outer calls helper + method call .saturating_add
        let callees: Vec<&str> = scan.calls.iter().map(|c| c.callee_ident.as_str()).collect();
        assert!(callees.contains(&"helper"), "{callees:?}");
        assert!(callees.contains(&"saturating_add"));

        // Features: one cfg row on `gated`
        assert!(scan.features.iter().any(|f| f.item_name == "gated" && f.cfg_expr.contains("experimental")));
    }

    #[test]
    fn captures_real_source_lines() {
        // Regression for the old `line = 0` bug: proc-macro2 span-locations
        // must yield the 1-based source line of each item name / call site.
        // The fixture's lib.rs starts with a newline, so `outer` is line 2.
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::create_dir_all(root.join("src")).unwrap();
        std::fs::write(root.join("Cargo.toml"), "[package]\nname = \"demo\"\nversion = \"0.1.0\"\nedition = \"2021\"\n").unwrap();
        std::fs::write(root.join("src/lib.rs"), r#"
pub fn outer(x: u32) -> u32 {
    helper(x) + 1
}
fn helper(x: u32) -> u32 { x.saturating_add(2) }

pub struct Foo { pub n: u32 }

#[cfg(feature = "experimental")]
pub fn gated() {}
"#).unwrap();

        let scan = scan_repo(root, "demo", Uuid::new_v4(), Utc::now()).unwrap();

        let line_of = |name: &str| scan.symbols.iter().find(|s| s.item_name == name).map(|s| s.line);
        assert_eq!(line_of("outer"), Some(2));
        assert_eq!(line_of("helper"), Some(5));
        assert_eq!(line_of("Foo"), Some(7));
        assert_eq!(line_of("gated"), Some(10));

        // Every symbol got a real (non-zero) line.
        assert!(scan.symbols.iter().all(|s| s.line > 0), "{:?}",
            scan.symbols.iter().map(|s| (&s.item_name, s.line)).collect::<Vec<_>>());

        // Call sites carry their own lines: `helper(x)` is on line 3.
        let helper_call = scan.calls.iter().find(|c| c.callee_ident == "helper").unwrap();
        assert_eq!(helper_call.line, 3);
        assert!(scan.calls.iter().all(|c| c.line > 0));

        // The cfg feature-gate row shares the gated fn's line.
        let gate = scan.features.iter().find(|f| f.item_name == "gated").unwrap();
        assert_eq!(gate.line, 10);
    }

    /// The multi-file PARALLEL path: a crate with many varied-size files must scan
    /// every file EXACTLY once (no drops, no dupes from the batched work-stealing +
    /// merge) and produce a stable result across runs. The single-file tests above
    /// only hit the serial fast-path; this drives `parse_files_parallel`'s LPT
    /// (heaviest-first) schedule, the atomic batch claims, and the original-order
    /// merge across all cores.
    #[test]
    fn parallel_scan_is_complete_and_stable() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::create_dir_all(root.join("src")).unwrap();
        std::fs::write(
            root.join("Cargo.toml"),
            "[package]\nname = \"demo\"\nversion = \"0.1.0\"\nedition = \"2021\"\n",
        )
        .unwrap();

        // One file per module, sizes varying wildly (file `i` carries `i` filler
        // fns + a unique marker), so the heaviest-first schedule has real skew to
        // balance and the work-stealing actually spreads the big files.
        const N: usize = 60;
        let mut lib_rs = String::new();
        for i in 0..N {
            lib_rs.push_str(&format!("pub mod f{i};\n"));
            let mut body = String::new();
            for j in 0..i {
                body.push_str(&format!("fn filler_{i}_{j}() {{ let _ = {j}; }}\n"));
            }
            body.push_str(&format!("pub fn marker_{i}() {{}}\n"));
            std::fs::write(root.join(format!("src/f{i}.rs")), body).unwrap();
        }
        std::fs::write(root.join("src/lib.rs"), lib_rs).unwrap();

        let scan1 = scan_repo(root, "demo", Uuid::new_v4(), Utc::now()).unwrap();

        // Completeness: every file's unique marker fn is present EXACTLY once
        // (proves no file was dropped, double-claimed, or merged twice).
        for i in 0..N {
            let hits = scan1
                .symbols
                .iter()
                .filter(|s| s.item_name == format!("marker_{i}"))
                .count();
            assert_eq!(hits, 1, "marker_{i} should appear exactly once, got {hits}");
        }
        let markers = scan1
            .symbols
            .iter()
            .filter(|s| s.item_name.starts_with("marker_"))
            .count();
        assert_eq!(markers, N, "expected {N} marker fns, got {markers}");

        // Stability: a second scan of the same tree yields the SAME multiset of
        // (file, name, line) symbol rows — the merge is deterministic, not racy.
        let scan2 = scan_repo(root, "demo", Uuid::new_v4(), Utc::now()).unwrap();
        let key = |s: &SymbolRow| (s.file.clone(), s.item_name.clone(), s.line);
        let mut a: Vec<_> = scan1.symbols.iter().map(key).collect();
        let mut b: Vec<_> = scan2.symbols.iter().map(key).collect();
        a.sort();
        b.sort();
        assert_eq!(a, b, "two scans of the same tree must produce identical rows");
    }

    /// The FLATTENED multi-repo scan (`scan_repos_flattened`) groups the union of
    /// every repo's files back per repo, and each member's rows are byte-identical
    /// to scanning that member alone — proving the one shared pool over the union
    /// doesn't cross-contaminate members or drop/dupe files.
    #[test]
    fn flattened_scan_groups_per_member_and_matches_solo() {
        fn mk(root: &Path, name: &str, body: &str) -> std::path::PathBuf {
            let dir = root.join(name);
            std::fs::create_dir_all(dir.join("src")).unwrap();
            std::fs::write(
                dir.join("Cargo.toml"),
                format!("[package]\nname = \"{name}\"\nversion = \"0.1.0\"\nedition = \"2021\"\n"),
            )
            .unwrap();
            std::fs::write(dir.join("src/lib.rs"), body).unwrap();
            dir
        }
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        // Three members of very different size, so the LPT schedule has real skew.
        let mut big = String::from("pub fn alpha_marker() {}\n");
        for i in 0..200 {
            big.push_str(&format!("fn filler_{i}() {{ let _ = {i}; }}\n"));
        }
        let a = mk(root, "alpha", &big);
        let b = mk(root, "beta", "pub fn beta_marker() {}\n#[test]\nfn beta_t() {}\n");
        let c = mk(root, "gamma", "pub struct GammaMarker;\n");

        let repos = vec![
            ("alpha".to_string(), a.clone()),
            ("beta".to_string(), b.clone()),
            ("gamma".to_string(), c.clone()),
        ];
        let snaps: Vec<Uuid> = (0..3).map(|_| Uuid::new_v4()).collect();
        let ts = Utc::now();
        let flat = scan_repos_flattened(&repos, &snaps, ts);

        assert_eq!(flat.len(), 3);
        // Each grouped scan carries ONLY its own member's rows + snapshot id.
        for (i, (name, _)) in repos.iter().enumerate() {
            assert_eq!(&flat[i].repo, name, "scan grouped to the right member");
            assert_eq!(flat[i].snapshot_id, snaps[i], "keeps the per-member snapshot id");
            assert!(
                flat[i].symbols.iter().all(|s| &s.crate_name == name),
                "member {name} must not carry another member's symbols"
            );
        }
        // Markers landed in the right member, exactly once.
        let has = |i: usize, n: &str| flat[i].symbols.iter().filter(|s| s.item_name == n).count();
        assert_eq!(has(0, "alpha_marker"), 1);
        assert_eq!(has(1, "beta_marker"), 1);
        assert_eq!(has(2, "GammaMarker"), 1);
        // The flattened beta scan equals a solo scan of beta (same symbol multiset).
        let solo_b = scan_repo(&b, "beta", snaps[1], ts).unwrap();
        let key = |s: &SymbolRow| (s.item_name.clone(), s.line);
        let mut f: Vec<_> = flat[1].symbols.iter().map(key).collect();
        let mut s: Vec<_> = solo_b.symbols.iter().map(key).collect();
        f.sort();
        s.sort();
        assert_eq!(f, s, "flattened member scan == solo scan of that member");
        // The test inventory rode along (build-free) for beta.
        assert!(flat[1].tests.iter().any(|t| t.test_name == "beta_t"));
    }

    /// MANUAL perf probe (not a correctness gate): scan the live nornir repo N
    /// times and print wall-time + symbol count. Run serial vs all-core via
    /// `NORNIR_SCAN_THREADS` to see the speedup / CPU saturation:
    ///   NORNIR_SCAN_THREADS=1 cargo test ... bench_scan_throughput_manual -- --ignored --nocapture
    ///   cargo test ...               bench_scan_throughput_manual -- --ignored --nocapture
    #[test]
    #[ignore = "manual perf probe, not a correctness gate"]
    fn bench_scan_throughput_manual() {
        let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
        let runs = 20;
        let t = std::time::Instant::now();
        let mut syms = 0usize;
        for _ in 0..runs {
            let scan = scan_repo(root, "nornir", Uuid::new_v4(), Utc::now()).unwrap();
            syms = scan.symbols.len();
        }
        let el = t.elapsed();
        eprintln!(
            "scan x{runs}: {:.2}s total | {:.1} ms/scan | {syms} symbols | workers={}",
            el.as_secs_f64(),
            el.as_secs_f64() * 1000.0 / runs as f64,
            super::scan_worker_budget(),
        );
    }

    /// PART A — the syn scan harvests the test INVENTORY (no build): a `#[test]`,
    /// a `#[tokio::test]` (async), and an `#[ignore]`-marked heavy test are all
    /// discovered with the correct `is_heavy`/`is_async`, while plain fns are not
    /// mistaken for tests. Inject a fixture crate, assert the harvested rows.
    #[test]
    fn scans_test_inventory_with_heavy_and_async_flags() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::create_dir_all(root.join("src")).unwrap();
        std::fs::write(
            root.join("Cargo.toml"),
            "[package]\nname = \"demo\"\nversion = \"0.1.0\"\nedition = \"2021\"\n",
        )
        .unwrap();
        std::fs::write(
            root.join("src/lib.rs"),
            r#"
pub fn not_a_test() {}

#[cfg(test)]
mod tests {
    #[test]
    fn light_one() { assert_eq!(1, 1); }

    #[tokio::test]
    async fn async_one() { assert!(true); }

    #[test]
    #[ignore]
    fn heavy_corpus() { assert!(true); }

    #[test_case(1 ; "case a")]
    fn macro_cased(x: u32) { let _ = x; }
}
"#,
        )
        .unwrap();

        let scan = scan_repo(root, "demo", Uuid::new_v4(), Utc::now()).unwrap();
        let by_name = |n: &str| scan.tests.iter().find(|t| t.test_name == n);

        // Exactly the four test fns are inventoried; the plain fn is NOT.
        let names: Vec<&str> = scan.tests.iter().map(|t| t.test_name.as_str()).collect();
        assert!(names.contains(&"light_one"), "{names:?}");
        assert!(names.contains(&"async_one"));
        assert!(names.contains(&"heavy_corpus"));
        assert!(names.contains(&"macro_cased"));
        assert!(!names.contains(&"not_a_test"), "plain fn must not be a test row");

        // Heavy gating flag: only the #[ignore] one is heavy.
        assert!(by_name("heavy_corpus").unwrap().is_heavy, "ignore → heavy");
        assert!(!by_name("light_one").unwrap().is_heavy);

        // Async flag: tokio::test and async fn → is_async; a plain #[test] is not.
        assert!(by_name("async_one").unwrap().is_async, "tokio::test → async");
        assert!(!by_name("light_one").unwrap().is_async);

        // Rows carry a real module path + line.
        let light = by_name("light_one").unwrap();
        assert!(light.module_path.contains("tests"), "module path: {}", light.module_path);
        assert!(light.line > 0);
        assert_eq!(light.crate_name, "demo");
    }
}