rdar 0.6.10

//! Parallel repository scan: walk + hash + extract in one pass, with
//! checkpointed caches so interrupted runs resume.
//!
//! Scanner invariants:
//! - **Monotonic resume**: the collector seeds from the previous cache and
//!   checkpoints a superset; deleted files are pruned only at the successful
//!   final save. Interrupting a resume run never loses prior progress.
//! - **Racy-mtime guard**: a file whose mtime is not strictly older than the
//!   cache's `saved_at` is never stat-trusted (git's racily-clean rule);
//!   `(0, 0)` mtimes are never trusted either.
//! - **Cache failures degrade**: an unwritable `.radar/` costs persistence,
//!   never the scan (warn once, keep going).
//! - Parse cache is keyed by `(lang, hash)` - identical bytes in different
//!   languages parse differently.

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::io;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{Receiver, SyncSender, sync_channel};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant, SystemTime};

use ignore::{WalkBuilder, WalkState};
use serde::Serialize;

use crate::cache::{FileEntry, ScanCache};
use crate::extract::{self, Extraction};
use crate::lang::Lang;
use crate::progress::{Counters, Spinner};

/// Directory names never worth scanning even when not gitignored
/// This mirrors the map-validation skip set.
const SKIP_DIRS: [&str; 16] = [
    ".git",
    ".radar",
    "target",
    "node_modules",
    "dist",
    "build",
    ".venv",
    "venv",
    "__pycache__",
    "vendor",
    "third_party",
    ".idea",
    ".vscode",
    ".cache",
    ".tox",
    "coverage",
];

/// Files larger than this are stat-tracked but not read or parsed.
const MAX_SOURCE_BYTES: u64 = 16 * 1024 * 1024;

/// Default checkpoint interval (files between incremental cache flushes).
const CHECKPOINT_EVERY: u64 = 2000;
/// Checkpoint at least this often while work is flowing (env-overridable for
/// deterministic tests: `RADAR_CHECKPOINT_SECS`).
const CHECKPOINT_SECS: u64 = 5;
const PARSE_SHARDS: usize = 64;

struct SharedExtraction {
    value: OnceLock<Arc<Extraction>>,
    reported: AtomicBool,
}

impl SharedExtraction {
    fn new() -> Self {
        Self {
            value: OnceLock::new(),
            reported: AtomicBool::new(false),
        }
    }
}

type ParseSlots = [Mutex<HashMap<(Lang, [u8; 32]), Arc<SharedExtraction>>>; PARSE_SHARDS];

#[derive(Debug, Default, Serialize)]
pub struct LangStats {
    pub files: u64,
    pub defs: u64,
    pub refs: u64,
}

#[derive(Debug, Default, Serialize)]
pub struct ScanStats {
    pub root: String,
    pub files_seen: u64,
    pub source_files: u64,
    pub parsed: u64,
    pub stat_hits: u64,
    pub hash_hits: u64,
    pub skipped_large: u64,
    pub defs: u64,
    pub refs: u64,
    pub per_lang: BTreeMap<&'static str, LangStats>,
    pub elapsed_ms: u64,
    pub git_repo: bool,
    /// False when `.radar/` could not be written (scan still valid).
    pub cache_persisted: bool,
}

pub struct ScanOpts {
    pub root: PathBuf,
    pub jobs: Option<usize>,
    pub quiet: bool,
}

enum Msg {
    /// Stats of an unchanged file - extraction reused from the old cache.
    StatHit { rel: String, entry: FileEntry },
    /// Content already known by (lang, hash) (rename/copy) - no parse.
    HashHit { rel: String, entry: FileEntry },
    /// Content first seen during this scan. Concurrent identical files share
    /// one extraction; `first` owns the parsed count and the rest are hash hits.
    Parsed {
        rel: String,
        entry: FileEntry,
        extraction: Arc<Extraction>,
        first: bool,
    },
    /// Non-source or oversized file - tracked for census only.
    Plain { rel: String, entry: FileEntry },
}

/// Run a full scan of `opts.root`, reusing and updating `.radar/state.bin`.
///
/// Errors with `ErrorKind::NotFound`/`NotADirectory` when the root is not a
/// readable directory (mapped to exit 2 by the CLI - never a side effect).
pub fn scan(opts: &ScanOpts) -> io::Result<ScanStats> {
    scan_full(opts).map(|(stats, _)| stats)
}

/// Like [`scan`], but also returns the final in-memory cache (used by
/// `radar map`/`check` so they never depend on cache persistence).
pub fn scan_full(opts: &ScanOpts) -> io::Result<(ScanStats, ScanCache)> {
    let started = Instant::now();
    let root = &opts.root;
    let meta = std::fs::metadata(root)?; // NotFound propagates untouched
    if !meta.is_dir() {
        return Err(io::Error::new(
            io::ErrorKind::NotADirectory,
            format!("{} is not a directory", root.display()),
        ));
    }
    crate::mcp::invalidate_query_snapshot(root);

    let old = Arc::new(ScanCache::load(root));
    let counters = Arc::new(Counters::default());
    let parse_slots: Arc<ParseSlots> =
        Arc::new(std::array::from_fn(|_| Mutex::new(HashMap::new())));
    let git_repo = root.join(".git").exists();
    // Resolve the ambiguous `.cls` extension once for the whole scan: Apex in a
    // Salesforce DX tree, unsupported otherwise.
    let salesforce = crate::lang::is_salesforce_project(root);

    let notice_printed = if !git_repo && !old.notices.no_git && !opts.quiet {
        eprintln!(
            "radar: no git detected - using radar's own change tracking; refresh stays O(diff)"
        );
        true
    } else {
        false
    };

    let spinner = Spinner::start("scanning", Arc::clone(&counters), opts.quiet);
    // Batched sends (fd's BatchSender pattern): one channel op per ~256
    // files instead of per file - the single collector's receive loop is
    // the serialization point on very large repos.
    let (tx, rx) = sync_channel::<Vec<Msg>>(64);

    // Collector seeds from the old cache (monotonic checkpoints) and prunes
    // deleted files only at the successful final save.
    let collector = {
        let old = Arc::clone(&old);
        let root = root.to_path_buf();
        let quiet = opts.quiet;
        let no_git_notice = old.notices.no_git || notice_printed;
        std::thread::spawn(move || collect(rx, &old, &root, no_git_notice, quiet))
    };

    // fd's cap: beyond 64 threads, startup overhead outweighs walk gains.
    let threads = opts
        .jobs
        .unwrap_or_else(|| std::thread::available_parallelism().map_or(4, |n| n.get().min(64)));
    let walker = WalkBuilder::new(root)
        .threads(threads.max(1))
        .filter_entry(|entry| {
            let name = entry.file_name().to_string_lossy();
            !(entry.file_type().is_some_and(|t| t.is_dir()) && SKIP_DIRS.contains(&name.as_ref()))
        })
        .build_parallel();

    walker.run(|| {
        let tx: SyncSender<Vec<Msg>> = tx.clone();
        let old = Arc::clone(&old);
        let counters = Arc::clone(&counters);
        let parse_slots = Arc::clone(&parse_slots);
        let root = root.to_path_buf();
        let batch: std::cell::RefCell<Vec<Msg>> = std::cell::RefCell::new(Vec::new());
        struct FlushOnDrop {
            tx: SyncSender<Vec<Msg>>,
            batch: std::cell::RefCell<Vec<Msg>>,
        }
        impl Drop for FlushOnDrop {
            fn drop(&mut self) {
                let pending = std::mem::take(&mut *self.batch.borrow_mut());
                if !pending.is_empty() {
                    let _ = self.tx.send(pending);
                }
            }
        }
        let flusher = FlushOnDrop { tx, batch };
        Box::new(move |entry| {
            let Ok(entry) = entry else {
                return WalkState::Continue;
            };
            if !entry.file_type().is_some_and(|t| t.is_file()) {
                return WalkState::Continue;
            }
            let path = entry.path();
            let Some(rel) = rel_path(&root, path) else {
                return WalkState::Continue;
            };
            let Ok(meta) = entry.metadata() else {
                return WalkState::Continue;
            };
            let mtime = mtime_of(meta.modified().ok());
            let size = meta.len();
            let ino = inode_of(&meta);
            counters.files.fetch_add(1, Ordering::Relaxed);

            // L1 stat fast-path - only when the signature matches AND the
            // mtime is strictly older than the last cache save (racy guard)
            // AND the mtime is a real timestamp.
            if let Some(prev) = old.files.get(&rel)
                && prev.mtime == mtime
                && prev.size == size
                && prev.ino == ino
                && mtime != (0, 0)
                && mtime < old.saved_at
                && (path.extension().and_then(|ext| ext.to_str()) != Some("cls")
                    || prev.lang == Lang::from_path(path, salesforce))
            {
                counters.cached.fetch_add(1, Ordering::Relaxed);
                let msg = Msg::StatHit {
                    rel,
                    entry: prev.clone(),
                };
                let mut b = flusher.batch.borrow_mut();
                b.push(msg);
                if b.len() >= 256 {
                    let out = std::mem::take(&mut *b);
                    drop(b);
                    if flusher.tx.send(out).is_err() {
                        return WalkState::Quit;
                    }
                }
                return WalkState::Continue;
            }

            if size > MAX_SOURCE_BYTES {
                let entry = FileEntry {
                    mtime,
                    size,
                    ino,
                    hash: [0u8; 32],
                    lang: None,
                };
                flusher.batch.borrow_mut().push(Msg::Plain { rel, entry });
                return WalkState::Continue;
            }

            let Ok(bytes) = std::fs::read(path) else {
                return WalkState::Continue;
            };
            let hash = *blake3::hash(&bytes).as_bytes();
            let lang = Lang::from_path(path, salesforce).or_else(|| Lang::from_shebang(&bytes));
            let entry = FileEntry {
                mtime,
                size,
                ino,
                hash,
                lang,
            };

            let msg = match lang {
                None => Msg::Plain { rel, entry },
                Some(lang) => {
                    if old.parses.contains_key(&(lang, hash)) {
                        counters.cached.fetch_add(1, Ordering::Relaxed);
                        Msg::HashHit { rel, entry }
                    } else {
                        let shard = hash[0] as usize % PARSE_SHARDS;
                        let shared = {
                            let mut slots = parse_slots[shard]
                                .lock()
                                .unwrap_or_else(|poisoned| poisoned.into_inner());
                            Arc::clone(
                                slots
                                    .entry((lang, hash))
                                    .or_insert_with(|| Arc::new(SharedExtraction::new())),
                            )
                        };
                        let extraction = Arc::clone(shared.value.get_or_init(|| {
                            let src = String::from_utf8_lossy(&bytes);
                            counters.parsed.fetch_add(1, Ordering::Relaxed);
                            Arc::new(extract::extract(lang, &src))
                        }));
                        let first = !shared.reported.swap(true, Ordering::AcqRel);
                        if !first {
                            counters.cached.fetch_add(1, Ordering::Relaxed);
                        }
                        Msg::Parsed {
                            rel,
                            entry,
                            extraction,
                            first,
                        }
                    }
                }
            };
            let mut b = flusher.batch.borrow_mut();
            b.push(msg);
            if b.len() >= 256 {
                let out = std::mem::take(&mut *b);
                drop(b);
                if flusher.tx.send(out).is_err() {
                    return WalkState::Quit;
                }
            }
            WalkState::Continue
        })
    });
    drop(parse_slots);
    drop(tx);

    let (cache, mut stats) = collector
        .join()
        .map_err(|_| io::Error::other("scan collector thread panicked"))?;

    spinner.finish();
    stats.root = root.display().to_string();
    stats.git_repo = git_repo;
    stats.elapsed_ms = started.elapsed().as_millis() as u64;

    // Per-language def/ref totals from the final cache.
    for entry in cache.files.values() {
        let Some(lang) = entry.lang else { continue };
        let Some(x) = cache.parses.get(&(lang, entry.hash)) else {
            continue;
        };
        let ls = stats.per_lang.entry(lang.name()).or_default();
        ls.files += 1;
        ls.defs += x.defs.len() as u64;
        ls.refs += x.refs.len() as u64;
        stats.defs += x.defs.len() as u64;
        stats.refs += x.refs.len() as u64;
    }
    let _ = crate::mcp::save_query_snapshot(root, &cache);
    Ok((stats, cache))
}

/// Collector: merges walked results over the previous cache (monotonic),
/// checkpoints periodically, prunes at the final save, and degrades - never
/// aborts - when the cache cannot be written.
fn collect(
    rx: Receiver<Vec<Msg>>,
    old: &ScanCache,
    root: &Path,
    no_git_notice: bool,
    quiet: bool,
) -> (ScanCache, ScanStats) {
    let checkpoint_every = std::env::var("RADAR_CHECKPOINT_EVERY")
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .unwrap_or(CHECKPOINT_EVERY);
    let checkpoint_secs = std::env::var("RADAR_CHECKPOINT_SECS")
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .unwrap_or(CHECKPOINT_SECS);
    let test_exit_after = std::env::var("RADAR_TEST_EXIT_AFTER")
        .ok()
        .and_then(|v| v.parse::<u64>().ok());

    // Seed from the previous cache: checkpoints are supersets (§7.6).
    let mut cache = old.clone();
    cache.notices.no_git = no_git_notice;
    let mut seen: BTreeSet<String> = BTreeSet::new();

    let mut stats = ScanStats {
        cache_persisted: true,
        ..ScanStats::default()
    };
    let mut save_warned = false;
    let mut since_checkpoint = 0u64;
    let mut last_checkpoint = Instant::now();
    let mut processed = 0u64;

    let try_save = |cache: &mut ScanCache, stats: &mut ScanStats, warned: &mut bool| {
        if let Err(e) = cache.save(root) {
            stats.cache_persisted = false;
            if !*warned {
                if !quiet {
                    eprintln!("radar: warning: cache not persisted ({e}); scan continues");
                }
                *warned = true;
            }
        }
    };

    for batch in rx {
        for msg in batch {
            stats.files_seen += 1;
            processed += 1;
            let (rel, entry) = match msg {
                Msg::StatHit { rel, entry } => {
                    stats.stat_hits += 1;
                    if let Some(lang) = entry.lang {
                        stats.source_files += 1;
                        if let Some(x) = old.parses.get(&(lang, entry.hash)) {
                            cache
                                .parses
                                .entry((lang, entry.hash))
                                .or_insert_with(|| x.clone());
                        }
                    }
                    (rel, entry)
                }
                Msg::HashHit { rel, entry } => {
                    stats.hash_hits += 1;
                    stats.source_files += 1;
                    if let Some(lang) = entry.lang
                        && let Some(x) = old.parses.get(&(lang, entry.hash))
                    {
                        cache
                            .parses
                            .entry((lang, entry.hash))
                            .or_insert_with(|| x.clone());
                    }
                    (rel, entry)
                }
                Msg::Parsed {
                    rel,
                    entry,
                    extraction,
                    first,
                } => {
                    if first {
                        stats.parsed += 1;
                    } else {
                        stats.hash_hits += 1;
                    }
                    stats.source_files += 1;
                    if let Some(lang) = entry.lang {
                        cache
                            .parses
                            .entry((lang, entry.hash))
                            .or_insert_with(|| (*extraction).clone());
                    }
                    (rel, entry)
                }
                Msg::Plain { rel, entry } => (rel, entry),
            };
            if entry.size > MAX_SOURCE_BYTES {
                stats.skipped_large += 1;
            }
            seen.insert(rel.clone());
            cache.files.insert(rel, entry);

            since_checkpoint += 1;
            if since_checkpoint >= checkpoint_every
                || last_checkpoint.elapsed() >= Duration::from_secs(checkpoint_secs)
            {
                try_save(&mut cache, &mut stats, &mut save_warned);
                since_checkpoint = 0;
                last_checkpoint = Instant::now();
            }

            // Hidden test hook: simulate a crash AFTER checkpoints but BEFORE
            // the final save, so resume tests are deterministic.
            if let Some(n) = test_exit_after
                && processed >= n
            {
                std::process::exit(crate::exit::TEST_SIMULATED_CRASH);
            }
        }
    }

    // Successful full walk: prune entries for files that no longer exist,
    // then garbage-collect unreferenced parses.
    cache.files.retain(|rel, _| seen.contains(rel));
    let referenced: BTreeSet<(Lang, [u8; 32])> = cache
        .files
        .values()
        .filter_map(|e| e.lang.map(|l| (l, e.hash)))
        .collect();
    cache.parses.retain(|k, _| referenced.contains(k));
    try_save(&mut cache, &mut stats, &mut save_warned);

    (cache, stats)
}

fn rel_path(root: &Path, path: &Path) -> Option<String> {
    let rel = path.strip_prefix(root).ok()?;
    let mut out = String::new();
    for part in rel.components() {
        if !out.is_empty() {
            out.push('/');
        }
        out.push_str(&part.as_os_str().to_string_lossy());
    }
    if out.is_empty() { None } else { Some(out) }
}

fn mtime_of(t: Option<SystemTime>) -> (u64, u32) {
    t.and_then(|t| t.duration_since(SystemTime::UNIX_EPOCH).ok())
        .map_or((0, 0), |d| (d.as_secs(), d.subsec_nanos()))
}

#[cfg(unix)]
fn inode_of(meta: &std::fs::Metadata) -> u64 {
    use std::os::unix::fs::MetadataExt;
    meta.ino()
}

#[cfg(not(unix))]
fn inode_of(_meta: &std::fs::Metadata) -> u64 {
    0
}