oo-ide 0.0.4

//! Async project-wide search used by the editor's Expanded search bar.
//!
//! This module is intentionally decoupled from in-file (Inline) search logic.
//! Results are streamed back in batches via [`Operation::SearchLocal`]
//! so the UI can display partial results immediately.
//!
//! # Parallelism
//!
//! The directory walk uses [`WalkBuilder::build_parallel`] so multiple worker
//! threads search files concurrently.  Each worker owns its own [`Searcher`]
//! (cheap to create) and shares the compiled [`RegexMatcher`] (thread-safe).
//!
//! # Batching
//!
//! Workers send individual results to an internal aggregator channel.  A
//! dedicated aggregator thread collects results and flushes them to the UI
//! channel every 100 ms (or when a batch-size cap is reached), keeping channel
//! traffic at ~10 sends/second regardless of match volume.
//!
//! # Cancellation
//!
//! The caller passes a [`CancellationToken`].  Workers return
//! [`WalkState::Quit`] when the token fires, which stops all parallel threads.
//! Inside a file the [`ResultSink`] checks on every matched line and returns
//! `Ok(false)` from [`Sink::matched`] to stop the current file.

use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc as std_mpsc;
use std::sync::Arc;
use std::time::{Duration, Instant};

use grep::matcher::Matcher;
use grep::regex::{RegexMatcher, RegexMatcherBuilder};
use grep::searcher::{Searcher, SearcherBuilder, Sink, SinkMatch};
use ignore::WalkBuilder;
use tokio_util::sync::CancellationToken;
use tokio::sync::mpsc::UnboundedSender;

use crate::operation::{MatchSpan, Operation, SearchOp};
use crate::views::editor::SearchOptions;

/// Flush interval for the aggregator thread.
const BATCH_FLUSH_INTERVAL: Duration = Duration::from_millis(100);
/// Maximum number of results to accumulate before forcing a flush regardless
/// of elapsed time.
const BATCH_SIZE_CAP: usize = 5000;

/// Walk the project tree in parallel, search every text file for `query`, and
/// stream results back via `tx` as [`SearchOp::AddProjectResult`] operations
/// batched into `Vec<Operation>` chunks.
///
/// Uses the `ignore` crate for gitignore-aware parallel walking and the `grep`
/// crate for fast buffered searching with automatic binary-file detection.
///
/// Designed to be called inside `tokio::task::spawn_blocking`.
#[allow(clippy::too_many_arguments)]
pub fn run_project_search(
    root: &Path,
    query: &str,
    opts: &SearchOptions,
    generation: u64,
    cancel: CancellationToken,
    tx: &UnboundedSender<Vec<Operation>>,
    gen_shared: &Arc<AtomicU64>,
    registry: Option<crate::file_index::SharedRegistry>,
) {
    if query.is_empty() {
        log::debug!("project_search gen={generation}: empty query, skipping");
        return;
    }
    if cancel.is_cancelled() {
        log::debug!("project_search gen={generation}: already cancelled before start");
        return;
    }
    log::debug!("project_search gen={generation}: starting, query={query:?}");

    let ignore_case =
        opts.ignore_case || (opts.smart_case && !query.chars().any(|c| c.is_uppercase()));

    let pattern = if opts.regex {
        query.to_owned()
    } else {
        regex::escape(query)
    };

    let matcher = match RegexMatcherBuilder::new()
        .case_insensitive(ignore_case)
        .build(&pattern)
    {
        Ok(m) => m,
        Err(_) => return,
    };

    let mut ob = ignore::overrides::OverrideBuilder::new(root);
    if !opts.include_glob.is_empty() && opts.include_glob != "*" {
        let _ = ob.add(&opts.include_glob);
    }
    if !opts.exclude_glob.is_empty() {
        let _ = ob.add(&format!("!{}", opts.exclude_glob));
    }
    let overrides = ob.build().unwrap_or_else(|_| ignore::overrides::Override::empty());

    let thread_count = std::thread::available_parallelism()
        .map(|n| n.get())
        .unwrap_or(4)
        .min(8);

    // Internal channel: workers → aggregator.  Workers send individual
    // (file, MatchSpan) pairs; the aggregator batches and flushes to `tx`.
    let (agg_tx, agg_rx) = std_mpsc::channel::<(PathBuf, MatchSpan)>();

    // Spawn the aggregator thread that batches results and flushes to the
    // UI channel every BATCH_FLUSH_INTERVAL.  Before each flush, the
    // aggregator checks the shared generation counter — if it has changed
    // (a new search was triggered), the buffered batch is stale and is
    // dropped instead of sent.
    let ui_tx = tx.clone();
    let agg_cancel = cancel.clone();
    let agg_gen = gen_shared.clone();
    let aggregator = std::thread::spawn(move || {
        let mut batch: Vec<Operation> = Vec::new();
        let mut last_flush = Instant::now();

        /// Flush `batch` to `ui_tx` only if `generation` still matches the
        /// shared counter — otherwise discard (the search was superseded).
        #[inline]
        fn maybe_flush(
            batch: &mut Vec<Operation>,
            ui_tx: &UnboundedSender<Vec<Operation>>,
            generation: u64,
            gen_shared: &AtomicU64,
        ) {
            if batch.is_empty() {
                return;
            }
            if gen_shared.load(Ordering::Relaxed) != generation {
                // Stale — discard without sending.
                batch.clear();
                return;
            }
            let _ = ui_tx.send(std::mem::take(batch));
        }

        loop {
            // Wait with a timeout so we flush periodically even if results
            // arrive slowly.
            match agg_rx.recv_timeout(BATCH_FLUSH_INTERVAL) {
                Ok((file, span)) => {
                    if agg_cancel.is_cancelled() {
                        break;
                    }
                    batch.push(Operation::SearchLocal(SearchOp::AddProjectResult {
                        file,
                        result: span,
                        generation,
                    }));
                    // Drain any additional results that are already available.
                    while let Ok((file, span)) = agg_rx.try_recv() {
                        batch.push(Operation::SearchLocal(SearchOp::AddProjectResult {
                            file,
                            result: span,
                            generation,
                        }));
                        if batch.len() >= BATCH_SIZE_CAP {
                            break;
                        }
                    }
                    if batch.len() >= BATCH_SIZE_CAP || last_flush.elapsed() >= BATCH_FLUSH_INTERVAL {
                        maybe_flush(&mut batch, &ui_tx, generation, &agg_gen);
                        last_flush = Instant::now();
                    }
                }
                Err(std_mpsc::RecvTimeoutError::Timeout) => {
                    maybe_flush(&mut batch, &ui_tx, generation, &agg_gen);
                    last_flush = Instant::now();
                    if agg_cancel.is_cancelled() {
                        break;
                    }
                }
                Err(std_mpsc::RecvTimeoutError::Disconnected) => {
                    // All senders dropped — walk is done.
                    break;
                }
            }
        }

        // Final flush of any remaining results.
        maybe_flush(&mut batch, &ui_tx, generation, &agg_gen);
    });

    // If a registry was provided and is ready, seed the candidate file list
    // from it (no filesystem walk). Otherwise fall back to the parallel
    // WalkBuilder path.
    let candidate_files_opt: Option<Vec<PathBuf>> = if let Some(shared) = registry {
        let guard = shared.load();
        (**guard).as_ref().map(|reg| reg.files_under(Path::new("")).into_iter().map(|rel| root.join(rel)).collect())
    } else {
        None
    };

    if let Some(candidate_files) = candidate_files_opt {
        // Seeded path: iterate candidate files (single-threaded) and feed the
        // same Searcher/Sink logic used by the walker. This avoids the cost of
        // a filesystem walk. (If desired this can be parallelized later.)
        let mut searcher = SearcherBuilder::new().line_number(true).build();
        for path in candidate_files {
            if cancel.is_cancelled() {
                break;
            }
            // Respect overrides (include/exclude globs) as a post-filter.
            let is_dir = path.metadata().map(|md| md.is_dir()).unwrap_or(false);
            if overrides.matched(&path, is_dir).is_ignore() {
                continue;
            }
            let sink = ResultSink {
                matcher: matcher.clone(),
                path: path.clone(),
                generation,
                cancel: cancel.clone(),
                agg_tx: agg_tx.clone(),
            };
            let _ = searcher.search_path(&matcher, &path, sink);
        }

        // Drop sender and wait for aggregator to flush results.
        drop(agg_tx);
        let _ = aggregator.join();
        log::debug!("project_search gen={generation}: seeded search complete");
    } else {
        // Fallback: perform the original parallel walk over the filesystem.
        let walker = WalkBuilder::new(root)
            .hidden(true)
            .git_ignore(true)
            .overrides(overrides)
            .threads(thread_count)
            .build_parallel();

        walker.run(|| {
            let matcher = matcher.clone();
            let cancel = cancel.clone();
            let agg_tx = agg_tx.clone();
            let mut searcher = SearcherBuilder::new().line_number(true).build();

            Box::new(move |entry| {
                if cancel.is_cancelled() {
                    return ignore::WalkState::Quit;
                }
                let entry = match entry {
                    Ok(e) => e,
                    Err(_) => return ignore::WalkState::Continue,
                };
                if entry.file_type().is_none_or(|ft| ft.is_dir()) {
                    return ignore::WalkState::Continue;
                }
                let path = entry.path().to_path_buf();
                let sink = ResultSink {
                    matcher: matcher.clone(),
                    path: path.clone(),
                    generation,
                    cancel: cancel.clone(),
                    agg_tx: agg_tx.clone(),
                };
                let _ = searcher.search_path(&matcher, entry.path(), sink);
                if cancel.is_cancelled() {
                    ignore::WalkState::Quit
                } else {
                    ignore::WalkState::Continue
                }
            })
        });

        // Drop the original sender so the aggregator sees Disconnected once all
        // worker-cloned senders are also dropped (which happens when `walker.run`
        // returns).
        drop(agg_tx);

        // Wait for the aggregator to finish its final flush.
        let _ = aggregator.join();
        log::debug!("project_search gen={generation}: walk complete");
    }
}

/// Receives matched lines from [`Searcher`] and sends individual match spans
/// to the aggregator channel for batching.
///
/// On each matched line every individual match span is extracted via the
/// matcher's `find_iter` so the UI can highlight precise byte ranges.
///
/// Returns `Ok(false)` when the cancellation token is set so the searcher
/// stops processing the current file as soon as possible.
struct ResultSink {
    matcher: RegexMatcher,
    path: PathBuf,
    generation: u64,
    cancel: CancellationToken,
    agg_tx: std_mpsc::Sender<(PathBuf, MatchSpan)>,
}

impl Sink for ResultSink {
    type Error = std::io::Error;

    fn matched(
        &mut self,
        _searcher: &Searcher,
        mat: &SinkMatch<'_>,
    ) -> Result<bool, Self::Error> {
        if self.cancel.is_cancelled() {
            log::debug!(
                "project_search gen={}: cancelled at line {} of {:?}, stopping file",
                self.generation,
                mat.line_number().unwrap_or(0),
                self.path,
            );
            return Ok(false);
        }

        // Line numbers from grep are 1-based; MatchSpan uses 0-based.
        let line_no = mat.line_number().unwrap_or(1).saturating_sub(1) as usize;
        let line_bytes = mat.bytes();

        // Strip the trailing line terminator for display purposes.
        let line_text = String::from_utf8_lossy(line_bytes)
            .trim_end_matches('\n').trim_end_matches('\r')
            .to_owned();

        // grep delivers one call per matching line; use the matcher's find_iter
        // to get every individual match span within that line.
        let cancel = &self.cancel;
        let agg_tx = &self.agg_tx;
        let path = &self.path;
        let generation = self.generation;
        let _ = self.matcher.find_iter(line_bytes, |m| {
            if cancel.is_cancelled() {
                log::debug!(
                    "project_search gen={generation}: cancelled inside find_iter at byte {} of {:?}",
                    m.start(),
                    path,
                );
                return false;
            }
            let span = MatchSpan {
                line: line_no,
                byte_start: m.start(),
                byte_end: m.end(),
                line_text: line_text.clone(),
            };
            // Send to the aggregator; if the channel is closed (aggregator
            // stopped due to cancellation) we just stop.
            if agg_tx.send((path.clone(), span)).is_err() {
                return false;
            }
            true
        });

        let still_alive = !self.cancel.is_cancelled();
        if !still_alive {
            log::debug!(
                "project_search gen={}: cancelled after find_iter on {:?}, stopping file",
                self.generation,
                self.path,
            );
        }
        Ok(still_alive)
    }
}


#[cfg(test)]
mod tests {
    use super::*;
    use tokio::sync::mpsc::unbounded_channel;
    use tokio_util::sync::CancellationToken;
    use crate::operation::{Operation, SearchOp};
    use crate::views::editor::SearchOptions;

    fn gen_shared(generation: u64) -> Arc<AtomicU64> {
        Arc::new(AtomicU64::new(generation))
    }

    /// Collect all operations drained from the receiver right now (non-blocking).
    fn drain_ops(rx: &mut tokio::sync::mpsc::UnboundedReceiver<Vec<Operation>>) -> Vec<Operation> {
        let mut out = Vec::new();
        while let Ok(batch) = rx.try_recv() {
            out.extend(batch);
        }
        out
    }

    fn add_result_ops(ops: &[Operation]) -> Vec<(std::path::PathBuf, usize, u64)> {
        ops.iter()
            .filter_map(|op| {
                if let Operation::SearchLocal(SearchOp::AddProjectResult {
                    file,
                    result,
                    generation,
                }) = op
                {
                    Some((file.clone(), result.line, *generation))
                } else {
                    None
                }
            })
            .collect()
    }

    #[test]
    fn finds_matches_across_multiple_files() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("a.txt"), "hello world\nfoo bar\nhello again").unwrap();
        std::fs::write(dir.path().join("b.txt"), "no match here\nhello there").unwrap();
        std::fs::write(dir.path().join("c.txt"), "nothing").unwrap();

        let (tx, mut rx) = unbounded_channel();
        run_project_search(
            dir.path(),
            "hello",
            &SearchOptions::default(),
            1,
            CancellationToken::new(),
            &tx,
            &gen_shared(1),
            None,
        );

        let ops = drain_ops(&mut rx);
        let results = add_result_ops(&ops);
        // All results must carry generation 1.
        assert!(results.iter().all(|(_, _, g)| *g == 1), "generation mismatch: {:?}", results);
        // Lines 0 and 2 in a.txt, line 1 in b.txt.
        let a_lines: Vec<usize> = results.iter().filter(|(f, _, _)| f.ends_with("a.txt")).map(|(_, l, _)| *l).collect();
        let b_lines: Vec<usize> = results.iter().filter(|(f, _, _)| f.ends_with("b.txt")).map(|(_, l, _)| *l).collect();
        assert!(a_lines.contains(&0), "expected match on line 0 of a.txt");
        assert!(a_lines.contains(&2), "expected match on line 2 of a.txt");
        assert!(b_lines.contains(&1), "expected match on line 1 of b.txt");
        // c.txt has no matches.
        assert!(results.iter().all(|(f, _, _)| !f.ends_with("c.txt")), "unexpected match in c.txt");
    }

    #[test]
    fn empty_query_produces_no_results() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("file.txt"), "some content").unwrap();

        let (tx, mut rx) = unbounded_channel();
        run_project_search(
            dir.path(),
            "",
            &SearchOptions::default(),
            1,
            CancellationToken::new(),
            &tx,
            &gen_shared(1),
            None,
        );

        assert!(drain_ops(&mut rx).is_empty(), "expected no ops for empty query");
    }

    #[test]
    fn pre_cancelled_token_produces_no_results() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("file.txt"), "hello world\nhello again").unwrap();

        let (tx, mut rx) = unbounded_channel();
        let cancel = CancellationToken::new();
        cancel.cancel(); // cancel before the search even starts

        run_project_search(dir.path(), "hello", &SearchOptions::default(), 1, cancel, &tx, &gen_shared(1), None);

        let ops = drain_ops(&mut rx);
        let results = add_result_ops(&ops);
        assert!(results.is_empty(), "expected no results after pre-cancellation, got {:?}", results);
    }

    #[test]
    fn ignore_case_option_finds_mixed_case() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("file.txt"), "Hello WORLD\nfoo\nhElLo World").unwrap();

        let (tx, mut rx) = unbounded_channel();
        let opts = SearchOptions { ignore_case: true, ..SearchOptions::default() };
        run_project_search(dir.path(), "hello", &opts, 1, CancellationToken::new(), &tx, &gen_shared(1), None);

        let ops = drain_ops(&mut rx);
        let results = add_result_ops(&ops);
        let lines: Vec<usize> = results.iter().map(|(_, l, _)| *l).collect();
        assert!(lines.contains(&0), "expected match on line 0");
        assert!(lines.contains(&2), "expected match on line 2");
        assert!(!lines.contains(&1), "did not expect match on line 1");
    }

    #[test]
    fn regex_option_matches_pattern() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("file.txt"), "abc123\nfoobar\nabc456").unwrap();

        let (tx, mut rx) = unbounded_channel();
        let opts = SearchOptions { regex: true, ..SearchOptions::default() };
        run_project_search(dir.path(), "abc\\d+", &opts, 1, CancellationToken::new(), &tx, &gen_shared(1), None);

        let ops = drain_ops(&mut rx);
        let results = add_result_ops(&ops);
        let lines: Vec<usize> = results.iter().map(|(_, l, _)| *l).collect();
        assert!(lines.contains(&0), "expected match on line 0");
        assert!(lines.contains(&2), "expected match on line 2");
        assert!(!lines.contains(&1), "did not expect match on line 1");
    }

    #[test]
    fn invalid_regex_produces_no_results() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("file.txt"), "abc").unwrap();

        let (tx, mut rx) = unbounded_channel();
        let opts = SearchOptions { regex: true, ..SearchOptions::default() };
        // "[invalid" is not a valid regex pattern.
        run_project_search(dir.path(), "[invalid", &opts, 1, CancellationToken::new(), &tx, &gen_shared(1), None);

        assert!(drain_ops(&mut rx).is_empty(), "expected no results for invalid regex");
    }

    #[test]
    fn generation_is_stamped_on_every_result() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("file.txt"), "match here\nalso match").unwrap();

        let (tx, mut rx) = unbounded_channel();
        run_project_search(
            dir.path(),
            "match",
            &SearchOptions::default(),
            42,
            CancellationToken::new(),
            &tx,
            &gen_shared(42),
            None,
        );

        let ops = drain_ops(&mut rx);
        let results = add_result_ops(&ops);
        assert!(!results.is_empty(), "expected at least one result");
        assert!(results.iter().all(|(_, _, g)| *g == 42), "all results must carry generation 42");
    }

    #[test]
    fn pre_cancelled_with_many_files_returns_quickly() {
        let dir = tempfile::tempdir().unwrap();
        for i in 0..300 {
            std::fs::write(
                dir.path().join(format!("{i:04}.txt")),
                "hello world",
            )
            .unwrap();
        }

        let (tx, mut rx) = unbounded_channel();
        let cancel = CancellationToken::new();
        cancel.cancel();

        let start = std::time::Instant::now();
        run_project_search(dir.path(), "hello", &SearchOptions::default(), 1, cancel, &tx, &gen_shared(1), None);
        let elapsed = start.elapsed();

        assert!(
            elapsed.as_millis() < 500,
            "pre-cancelled search with 300 files took too long: {elapsed:?}"
        );
        assert!(
            drain_ops(&mut rx).is_empty(),
            "expected no results after pre-cancellation"
        );
    }

    #[test]
    fn cancel_mid_search_terminates_and_emits_partial_results() {
        let dir = tempfile::tempdir().unwrap();
        // Create enough files to make the parallel search take a measurable
        // amount of time without being so few that it finishes before cancel
        // fires.  With the parallel walker small files are processed very
        // quickly, so we need a large number.
        let file_count = 2000;
        let matches_per_file = 2;
        for i in 0..file_count {
            std::fs::write(
                dir.path().join(format!("{i:04}.txt")),
                "needle in a haystack\nanother needle\nno match here",
            )
            .unwrap();
        }

        let (tx, mut rx) = unbounded_channel();
        let cancel = CancellationToken::new();
        let cancel_clone = cancel.clone();
        let root = dir.path().to_path_buf();

        // Run the search on a background OS thread so we can cancel from here.
        let gen_s = gen_shared(1);
        let search_thread = std::thread::spawn(move || {
            run_project_search(
                &root,
                "needle",
                &SearchOptions::default(),
                1,
                cancel_clone,
                &tx,
                &gen_s,
                None,
            );
        });

        // Give the search thread a moment to start processing, then cancel.
        std::thread::sleep(std::time::Duration::from_millis(5));
        cancel.cancel();

        // The search thread must finish well within a generous timeout; it
        // should stop at the next file-boundary after the cancel fires.
        let start = std::time::Instant::now();
        search_thread.join().expect("search thread panicked");
        assert!(
            start.elapsed().as_secs() < 10,
            "search thread did not stop within 10 s after cancellation"
        );

        // Drain whatever partial results arrived before the cancel; we only
        // require that the total is strictly less than the maximum possible
        // (file_count × matches_per_file).  If cancel fired very early we may
        // have 0 results, which is also fine.
        let max_results = file_count * matches_per_file;
        let total = drain_ops(&mut rx).len();
        assert!(
            total < max_results,
            "expected fewer than {max_results} results after cancellation, got {total}"
        );
    }

    #[test]
    fn byte_spans_are_correct_for_multiple_matches_per_line() {
        let dir = tempfile::tempdir().unwrap();
        // "hi" appears at bytes 0-2 and 3-5 on the first line.
        std::fs::write(dir.path().join("file.txt"), "hihihi\nno match").unwrap();

        let (tx, mut rx) = unbounded_channel();
        run_project_search(
            dir.path(),
            "hi",
            &SearchOptions::default(),
            1,
            CancellationToken::new(),
            &tx,
            &gen_shared(1),
            None,
        );

        let ops = drain_ops(&mut rx);
        let spans: Vec<(usize, usize, usize)> = ops
            .iter()
            .filter_map(|op| {
                if let Operation::SearchLocal(SearchOp::AddProjectResult { result, .. }) = op {
                    Some((result.line, result.byte_start, result.byte_end))
                } else {
                    None
                }
            })
            .collect();

        // Expect 3 non-overlapping matches on line 0.
        let line0: Vec<_> = spans.iter().filter(|(l, _, _)| *l == 0).collect();
        assert_eq!(line0.len(), 3, "expected 3 matches on line 0, got {line0:?}");
        // Spans must be [0,2), [2,4), [4,6)
        assert!(line0.contains(&&(0, 0, 2)), "missing span (0,0,2)");
        assert!(line0.contains(&&(0, 2, 4)), "missing span (0,2,4)");
        assert!(line0.contains(&&(0, 4, 6)), "missing span (0,4,6)");
    }

    #[test]
    fn stale_generation_suppresses_results() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("file.txt"), "hello world\nhello again").unwrap();

        let (tx, mut rx) = unbounded_channel();
        // Pass generation 1 as the search generation, but set the shared
        // atomic to 2 — simulating the user having typed a new query while
        // this search was in flight.
        let shared = Arc::new(AtomicU64::new(2));
        run_project_search(
            dir.path(),
            "hello",
            &SearchOptions::default(),
            1,
            CancellationToken::new(),
            &tx,
            &shared,
            None,
        );

        let ops = drain_ops(&mut rx);
        assert!(ops.is_empty(), "stale generation should suppress all results, got {ops:?}");
    }
}