skim 5.7.0

//! This module contains the matching coordinator
use crate::thread_pool::{self, ThreadPool};
use crate::tui::item_list::{MergeStrategy, ProcessedItems};
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

use crate::engine::normalized::NormalizedEngineFactory;
use crate::engine::split::SplitMatchEngineFactory;
use crate::item::{ItemPool, MatchedItem, RankBuilder};
use crate::prelude::{AndOrEngineFactory, ExactOrFuzzyEngineFactory, RegexEngineFactory};
use crate::{CaseMatching, MatchEngineFactory, SkimItem, SkimOptions};

/// Merges per-worker match results and writes them into `processed_items`.
///
/// When `no_sort` is false, concatenates the pre-sorted worker results into a
/// single contiguous `Vec` and calls `sort()`.  Rust's stable sort (driftsort
/// since 1.81, a `TimSort` variant before that) detects the k pre-sorted runs
/// and merges them in O(n log k) on contiguous memory — benchmarking shows
/// this consistently outperforms tree-based or fold-based merge strategies
/// due to driftsort's cache-friendly single-buffer merge passes.
///
/// When `no_sort` is true, worker results arrive in chunk-index order and are
/// flattened without sorting.
///
/// Signals `needs_render` after writing so the UI picks up the new data.
fn input_index(tac: bool, start: usize, batch_len: usize, batch_index: usize) -> usize {
    debug_assert!(batch_index < batch_len);
    if tac {
        start + batch_len - 1 - batch_index
    } else {
        start + batch_index
    }
}

fn merge_worker_results(
    worker_results: Vec<Vec<MatchedItem>>,
    no_sort: bool,
    processed_items: &Mutex<Option<ProcessedItems>>,
    merge_strategy: MergeStrategy,
    generation: usize,
    current_generation: &AtomicUsize,
    needs_render: &AtomicBool,
) {
    if current_generation.load(Ordering::Acquire) != generation {
        return;
    }

    let total_len: usize = worker_results.iter().map(Vec::len).sum();
    let mut items = Vec::with_capacity(total_len);
    for chunk in worker_results {
        items.extend(chunk);
    }

    if !no_sort {
        // Each worker's sub-list is already sorted by `prepare`, so stable
        // sort detects the pre-existing runs and merges them efficiently.
        items.sort();
    }

    trace!("matcher stop, total matched: {}", items.len());

    // Validate while holding the result lock so an old matcher cannot overwrite
    // results that belong to a newer query generation.
    let mut guard = processed_items
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    if current_generation.load(Ordering::Acquire) != generation {
        return;
    }
    if matches!(merge_strategy, MergeStrategy::Replace) {
        *guard = Some(ProcessedItems {
            items,
            merge: MergeStrategy::Replace,
            generation,
        });
        drop(guard);
        needs_render.store(true, Ordering::Relaxed);
        return;
    }
    match &mut *guard {
        Some(existing) if existing.generation != generation => {
            *guard = Some(ProcessedItems {
                items,
                merge: MergeStrategy::Replace,
                generation,
            });
        }
        Some(existing) => {
            if no_sort {
                if matches!(merge_strategy, MergeStrategy::Prepend) {
                    items.append(&mut existing.items);
                    existing.items = items;
                } else {
                    existing.items.extend(items);
                }
            } else {
                // Both sides are fully sorted — one O(n+m) merge.
                MatchedItem::merge_into_sorted(&mut existing.items, items);
            }
        }
        None => {
            *guard = Some(ProcessedItems {
                items,
                merge: merge_strategy,
                generation,
            });
        }
    }
    // Guard is dropped here, releasing the lock before we signal the render flag.

    needs_render.store(true, Ordering::Relaxed);
}

//==============================================================================
/// Control handle for a running matcher operation.
///
/// Provides methods to check status, retrieve results, and stop the matcher.
pub struct MatcherControl {
    stopped: Arc<AtomicBool>,
    interrupt: Arc<AtomicBool>,
    processed: Arc<AtomicUsize>,
    matched: Arc<AtomicUsize>,
}

impl Default for MatcherControl {
    fn default() -> Self {
        Self {
            stopped: Arc::new(AtomicBool::new(true)),
            interrupt: Arc::new(AtomicBool::new(false)),
            processed: Default::default(),
            matched: Default::default(),
        }
    }
}

impl MatcherControl {
    /// Returns the number of items that have been processed so far.
    #[must_use]
    pub fn get_num_processed(&self) -> usize {
        self.processed.load(Ordering::Relaxed)
    }

    /// Returns the number of items that have matched so far.
    #[must_use]
    pub fn get_num_matched(&self) -> usize {
        self.matched.load(Ordering::Relaxed)
    }

    /// Signals the matcher to stop processing.
    pub fn kill(&mut self) {
        self.interrupt.store(true, Ordering::Relaxed);
    }

    /// Returns true if the matcher has stopped (either completed or killed).
    #[must_use]
    pub fn stopped(&self) -> bool {
        self.stopped.load(Ordering::Relaxed)
    }
}

impl Drop for MatcherControl {
    fn drop(&mut self) {
        self.kill();
    }
}

//==============================================================================
/// The main matcher that coordinates fuzzy/exact matching of items against a query.
pub struct Matcher {
    engine_factory: Rc<dyn MatchEngineFactory>,
    case_matching: CaseMatching,
    /// The rank builder shared with all engines; used to attach criteria to `MatchedItem`s.
    pub rank_builder: Arc<RankBuilder>,
}

impl Matcher {
    /// Creates a new Matcher builder with the given engine factory.
    pub fn builder(engine_factory: Rc<dyn MatchEngineFactory>) -> Self {
        Self {
            engine_factory,
            case_matching: CaseMatching::default(),
            rank_builder: Arc::new(RankBuilder::default()),
        }
    }

    /// Sets the case matching mode (smart, ignore, or respect).
    #[must_use]
    pub fn case(mut self, case_matching: CaseMatching) -> Self {
        self.case_matching = case_matching;
        self
    }

    /// Sets the rank builder (carries tiebreak criteria).
    #[must_use]
    pub fn rank_builder(mut self, rank_builder: Arc<RankBuilder>) -> Self {
        self.rank_builder = rank_builder;
        self
    }

    /// Finalizes the builder and returns the configured Matcher.
    #[must_use]
    pub fn build(self) -> Self {
        self
    }

    /// Creates a `MatchEngineFactory` from the given options.
    ///
    /// This is useful when you need the factory directly (e.g., for filter mode)
    /// without creating a full Matcher instance.
    #[must_use]
    pub fn create_engine_factory(options: &SkimOptions) -> Rc<dyn MatchEngineFactory> {
        Self::create_engine_factory_with_builder(options).0
    }

    /// Creates a `MatchEngineFactory` and the associated `RankBuilder` from the given options.
    ///
    /// Returns both so callers can attach the builder to `MatchedItem`s for lazy sort-key
    /// computation.
    #[must_use]
    pub fn create_engine_factory_with_builder(options: &SkimOptions) -> (Rc<dyn MatchEngineFactory>, Arc<RankBuilder>) {
        if options.regex {
            let rank_builder = Arc::new(RankBuilder::new(options.tiebreak.clone()).tac(options.tac));
            let regex_factory = RegexEngineFactory::builder().rank_builder(rank_builder.clone());
            let factory: Rc<dyn MatchEngineFactory> = if options.normalize {
                Rc::new(NormalizedEngineFactory::new(regex_factory))
            } else {
                Rc::new(regex_factory)
            };
            (factory, rank_builder)
        } else {
            let rank_builder = Arc::new(RankBuilder::new(options.tiebreak.clone()).tac(options.tac));
            log::debug!("Creating matcher for algo {:?}", options.algorithm);
            let fuzzy_engine_factory = ExactOrFuzzyEngineFactory::builder()
                .fuzzy_algorithm(options.algorithm)
                .exact_mode(options.exact)
                .typos(options.typos)
                .filter_mode(options.filter.is_some())
                .last_match(options.last_match)
                .rank_builder(rank_builder.clone())
                .build();

            let mut factory: Box<dyn MatchEngineFactory> = Box::new(fuzzy_engine_factory);

            // If split_match is enabled, wrap the fuzzy factory with SplitMatchEngineFactory
            if let Some(delimiter) = options.split_match {
                factory = Box::new(SplitMatchEngineFactory::new(factory, delimiter));
            }

            // Wrap with AndOrEngineFactory so that queries like "foo:bar baz:qux" work
            factory = Box::new(AndOrEngineFactory::new(factory));

            // Wrap with NormalizedEngineFactory if normalization is requested
            if options.normalize {
                factory = Box::new(NormalizedEngineFactory::new(factory));
            }

            let factory: Rc<dyn MatchEngineFactory> = Rc::new(factory);
            (factory, rank_builder)
        }
    }

    /// Creates a Matcher configured from the given `SkimOptions`.
    #[must_use]
    pub fn from_options(options: &SkimOptions) -> Self {
        let (engine_factory, rank_builder) = Self::create_engine_factory_with_builder(options);
        Matcher::builder(engine_factory)
            .case(options.case)
            .rank_builder(rank_builder)
            .build()
    }

    /// Returns the case matching setting for this matcher.
    #[must_use]
    pub fn case_matching(&self) -> CaseMatching {
        self.case_matching
    }

    /// Returns a reference to the engine factory.
    #[must_use]
    pub fn engine_factory(&self) -> &Rc<dyn MatchEngineFactory> {
        &self.engine_factory
    }

    /// Runs the matcher on items from the pool in a background thread.
    ///
    /// When matching completes, the coordinator merges results directly into
    /// `processed_items` according to `merge_strategy`, then signals
    /// `needs_render` so the UI picks up the new data on its next tick.
    ///
    /// Returns a `MatcherControl` that can be used to monitor progress or
    /// stop the matcher.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn run(
        &self,
        query: &str,
        item_pool: &Arc<ItemPool>,
        thread_pool: &Arc<ThreadPool>,
        processed_items: Arc<Mutex<Option<ProcessedItems>>>,
        merge_strategy: MergeStrategy,
        no_sort: bool,
        tac: bool,
        generation: usize,
        current_generation: Arc<AtomicUsize>,
        needs_render: Arc<AtomicBool>,
    ) -> MatcherControl {
        let matcher_engine = self.engine_factory.create_engine_with_case(query, self.case_matching);
        debug!("engine: {matcher_engine}");
        let stopped = Arc::new(AtomicBool::new(false));
        let stopped_clone = stopped.clone();
        let interrupt = Arc::new(AtomicBool::new(false));
        let interrupt_clone = interrupt.clone();
        let processed = Arc::new(AtomicUsize::new(0));
        let processed_clone = processed.clone();
        let matched = Arc::new(AtomicUsize::new(0));
        let matched_clone = matched.clone();
        let rank_builder = self.rank_builder.clone();

        // Take items synchronously before spawning to avoid a race condition:
        // if we took items inside the spawned closure, a subsequent restart_matcher()
        // could call kill() + reset() before the old closure runs, causing the old
        // closure to re-take items that should belong to the new matcher.
        let start = item_pool.num_taken();
        let items = item_pool.take();
        let total = items.len();
        trace!("matcher start, total: {total}");

        // The coordinator runs on a dedicated OS thread so it does not occupy
        // a pool slot while waiting for workers.  All pool threads are
        // therefore available for the parallel matching work.
        let num_workers = thread_pool.num_threads();
        let pool_for_work = Arc::clone(thread_pool);

        std::thread::spawn(move || {
            // Process items in parallel using a shared work queue.  Each worker
            // thread atomically grabs the next available chunk, processes it,
            // and immediately merges its partial results.  This means threads
            // that finish early automatically pick up more work, providing
            // natural load balancing.
            //
            // The chunk size controls the granularity of work distribution and
            // the frequency of atomic counter updates / interrupt checks.
            const CHUNK_SIZE: usize = 1 << 12;

            // Convert items into an Arc slice so all workers can share them.
            let shared_items: Arc<[Arc<dyn SkimItem>]> = items.into();

            // Clones for the process_chunk closure.
            let matcher_engine: Arc<dyn crate::MatchEngine> = Arc::from(matcher_engine);
            let interrupt_for_work = Arc::clone(&interrupt);
            let processed_for_work = Arc::clone(&processed);
            let matched_for_work = Arc::clone(&matched);
            let rank_builder_for_work = Arc::clone(&rank_builder);

            thread_pool::parallel_work_queue(
                &pool_for_work,
                num_workers,
                &shared_items,
                CHUNK_SIZE,
                no_sort,
                // identity – seed value for each worker's local accumulator
                Vec::<MatchedItem>::new,
                // process_chunk – called for each chunk; returns a Vec of matches
                move |chunk_start, chunk: &[Arc<dyn crate::SkimItem>]| {
                    // Check interrupt before processing this chunk.
                    if interrupt_for_work.load(Ordering::Relaxed) {
                        return Vec::new();
                    }

                    let mut local_matches = Vec::new();
                    let mut chunk_matched: usize = 0;

                    for (i, item) in chunk.iter().enumerate() {
                        if let Some(match_result) = matcher_engine.match_item(item.as_ref()) {
                            chunk_matched += 1;
                            let mut rank = match_result.rank;
                            let batch_index = chunk_start + i;
                            // `take()` reverses each tac batch, so recover the
                            // item's stable ordinal in the original input stream.
                            let index = input_index(tac, start, total, batch_index);
                            rank.index = i32::try_from(index).unwrap_or(i32::MAX);
                            local_matches.push(MatchedItem::new(
                                Arc::clone(item),
                                rank,
                                Some(match_result.matched_range),
                                &rank_builder_for_work,
                            ));
                        }
                    }

                    // Flush counters for this chunk so the UI sees progress.
                    processed_for_work.fetch_add(chunk.len(), Ordering::Relaxed);
                    if chunk_matched > 0 {
                        matched_for_work.fetch_add(chunk_matched, Ordering::Relaxed);
                    }

                    local_matches
                },
                // reduce – accumulate chunk matches into the worker-local Vec.
                // No sorting here — that would be O(m²/chunk_size) per worker.
                |acc: &mut Vec<MatchedItem>, mut partial: Vec<MatchedItem>| {
                    if acc.len() >= partial.len() {
                        acc.extend(partial);
                    } else {
                        partial.append(acc);
                        *acc = partial;
                    }
                },
                // prepare – sort each worker's accumulator **on the worker
                // thread** so that sorting runs in parallel across all workers.
                // A single O((m/k)·log(m/k)) sort per worker is far cheaper
                // than sorting during reduce.
                // sort_unstable is used here because the worker's accumulator
                // has no pre-existing sorted runs (items were appended in
                // chunk order), so driftsort's run-detection overhead is pure
                // cost.  The final merge uses sort() so that driftsort can
                // exploit the k sorted runs produced by the workers.
                |acc: &mut Vec<MatchedItem>| acc.sort_unstable(),
                // merge – concat pre-sorted worker results and sort().
                // Rust's stable sort detects the k sorted runs and merges
                // them in O(n log k), then writes into processed_items.
                |worker_results: Vec<Vec<MatchedItem>>| {
                    if interrupt.load(Ordering::SeqCst) {
                        return;
                    }

                    merge_worker_results(
                        worker_results,
                        no_sort,
                        &processed_items,
                        merge_strategy,
                        generation,
                        &current_generation,
                        &needs_render,
                    );
                },
            );
            stopped.store(true, Ordering::Relaxed);
        });

        MatcherControl {
            stopped: stopped_clone,
            interrupt: interrupt_clone,
            matched: matched_clone,
            processed: processed_clone,
        }
    }
}

#[cfg(test)]
#[cfg_attr(coverage, coverage(off))]
mod tests {
    use super::*;
    use crate::Rank;
    use crate::item::RankBuilder;
    use crate::options::SkimOptionsBuilder;

    fn matched(text: &str, index: i32) -> MatchedItem {
        MatchedItem::new(
            Arc::new(text.to_string()),
            Rank {
                index,
                ..Default::default()
            },
            None,
            &RankBuilder::default(),
        )
    }

    #[test]
    fn from_options_exposes_case_and_factory() {
        let options = SkimOptionsBuilder::default()
            .case(CaseMatching::Ignore)
            .build()
            .unwrap();
        let matcher = Matcher::from_options(&options);
        assert_eq!(matcher.case_matching(), CaseMatching::Ignore);
        // The factory builds a working engine.
        let engine = matcher.engine_factory().create_engine("foo");
        assert!(engine.match_item(&"foobar".to_string()).is_some());
    }

    #[test]
    fn create_engine_factory_builds_fuzzy_engine() {
        let options = SkimOptions::default();
        let factory = Matcher::create_engine_factory(&options);
        let engine = factory.create_engine("fb");
        assert!(engine.match_item(&"foobar".to_string()).is_some());
    }

    #[test]
    fn create_engine_factory_regex_with_normalize() {
        let options = SkimOptionsBuilder::default()
            .regex(true)
            .normalize(true)
            .build()
            .unwrap();
        let (factory, _rank) = Matcher::create_engine_factory_with_builder(&options);
        let engine = factory.create_engine("ba.");
        assert!(engine.match_item(&"foobar".to_string()).is_some());
    }

    #[test]
    fn regex_factory_uses_configured_tiebreak() {
        let options = SkimOptionsBuilder::default()
            .regex(true)
            .tiebreak(vec![crate::RankCriteria::Length])
            .build()
            .unwrap();
        let (factory, rank_builder) = Matcher::create_engine_factory_with_builder(&options);
        let engine = factory.create_engine("a");
        let mut matches: Vec<_> = ["aaaa", "a", "aaa"]
            .into_iter()
            .enumerate()
            .map(|(index, text)| {
                let item: Arc<dyn SkimItem> = Arc::new(text.to_string());
                let mut result = engine.match_item(item.as_ref()).unwrap();
                result.rank.index = i32::try_from(index).unwrap();
                MatchedItem::new(item, result.rank, Some(result.matched_range), &rank_builder)
            })
            .collect();
        matches.sort();

        let output: Vec<_> = matches.iter().map(|item| item.text().into_owned()).collect();
        assert_eq!(output, ["a", "aaa", "aaaa"]);
    }

    fn merge_test_results(
        worker_results: Vec<Vec<MatchedItem>>,
        no_sort: bool,
        processed_items: &Mutex<Option<ProcessedItems>>,
        merge_strategy: MergeStrategy,
        needs_render: &AtomicBool,
    ) {
        let generation = AtomicUsize::new(0);
        merge_worker_results(
            worker_results,
            no_sort,
            processed_items,
            merge_strategy,
            0,
            &generation,
            needs_render,
        );
    }

    #[test]
    fn merge_worker_results_replace_sorts() {
        let processed = Mutex::new(None);
        let needs_render = AtomicBool::new(false);
        let workers = vec![vec![matched("b", 1)], vec![matched("a", 0)]];
        merge_test_results(workers, false, &processed, MergeStrategy::Replace, &needs_render);

        assert!(needs_render.load(Ordering::Relaxed));
        let guard = processed.lock().unwrap();
        let items = &guard.as_ref().unwrap().items;
        assert_eq!(items.len(), 2);
    }

    #[test]
    fn stale_generation_cannot_publish_results() {
        let processed = Mutex::new(None);
        let needs_render = AtomicBool::new(false);
        let generation = AtomicUsize::new(2);

        merge_worker_results(
            vec![vec![matched("stale", 0)]],
            false,
            &processed,
            MergeStrategy::Replace,
            1,
            &generation,
            &needs_render,
        );

        assert!(processed.lock().unwrap().is_none());
        assert!(!needs_render.load(Ordering::Relaxed));
    }

    #[test]
    fn current_results_replace_pending_stale_generation() {
        let processed = Mutex::new(Some(ProcessedItems {
            items: vec![matched("stale", 0)],
            merge: MergeStrategy::SortedMerge,
            generation: 0,
        }));
        let needs_render = AtomicBool::new(false);
        let generation = AtomicUsize::new(1);

        merge_worker_results(
            vec![vec![matched("current", 1)]],
            false,
            &processed,
            MergeStrategy::SortedMerge,
            1,
            &generation,
            &needs_render,
        );

        let guard = processed.lock().unwrap();
        let result = guard.as_ref().unwrap();
        assert_eq!(result.generation, 1);
        assert_eq!(result.items.len(), 1);
        assert_eq!(result.items[0].text(), "current");
        assert!(matches!(result.merge, MergeStrategy::Replace));
    }

    #[test]
    fn merge_worker_results_no_sort_preserves_chunk_order() {
        let processed = Mutex::new(None);
        let needs_render = AtomicBool::new(false);
        let workers = vec![
            vec![matched("a", 0), matched("b", 1)],
            vec![matched("c", 2), matched("d", 3)],
            vec![matched("e", 4), matched("f", 5)],
        ];
        merge_test_results(workers, true, &processed, MergeStrategy::Replace, &needs_render);

        let guard = processed.lock().unwrap();
        let items = &guard.as_ref().unwrap().items;
        let indexes: Vec<i32> = items.iter().map(|item| item.rank.index).collect();
        assert_eq!(indexes, vec![0, 1, 2, 3, 4, 5]);
    }

    #[test]
    fn merge_worker_results_append_no_sort_extends_existing() {
        let processed = Mutex::new(None);
        let needs_render = AtomicBool::new(false);

        // First append establishes the existing list.
        merge_test_results(
            vec![vec![matched("a", 0)]],
            true,
            &processed,
            MergeStrategy::Append,
            &needs_render,
        );
        // Second append with no_sort extends the existing list in place.
        merge_test_results(
            vec![vec![matched("b", 1)]],
            true,
            &processed,
            MergeStrategy::Append,
            &needs_render,
        );

        let guard = processed.lock().unwrap();
        assert_eq!(guard.as_ref().unwrap().items.len(), 2);
    }

    #[test]
    fn tac_input_index_spans_incremental_batches() {
        let first_batch: Vec<_> = (0..3).map(|index| input_index(true, 0, 3, index)).collect();
        let second_batch: Vec<_> = (0..2).map(|index| input_index(true, 3, 2, index)).collect();

        assert_eq!(first_batch, [2, 1, 0]);
        assert_eq!(second_batch, [4, 3]);
        assert_eq!(input_index(false, 3, 2, 0), 3);
        assert_eq!(input_index(false, 3, 2, 1), 4);
    }

    #[test]
    fn merge_worker_results_prepend_no_sort_places_new_batch_first() {
        let processed = Mutex::new(None);
        let needs_render = AtomicBool::new(false);

        merge_test_results(
            vec![vec![matched("c", 2), matched("b", 1), matched("a", 0)]],
            true,
            &processed,
            MergeStrategy::Prepend,
            &needs_render,
        );
        merge_test_results(
            vec![vec![matched("e", 4), matched("d", 3)]],
            true,
            &processed,
            MergeStrategy::Prepend,
            &needs_render,
        );

        let guard = processed.lock().unwrap();
        let indexes: Vec<i32> = guard
            .as_ref()
            .unwrap()
            .items
            .iter()
            .map(|item| item.rank.index)
            .collect();
        assert_eq!(indexes, [4, 3, 2, 1, 0]);
    }
}