sakurs-core 0.1.2

use std::sync::Arc;

use rayon::prelude::*;

use crate::{
    application::{
        chunking::{ChunkManager, TextChunk},
        config::{ProcessingError, ProcessingResult, ProcessorConfig},
        parser::TextParser,
    },
    domain::{
        prefix_sum::{ChunkStartState, PrefixSumComputer},
        reduce::BoundaryReducer,
        types::{Boundary, DeltaEntry, DeltaVec, PartialState},
    },
    LanguageRules,
};

use super::execution_mode::ExecutionMode;

/// Result of delta-stack processing with metadata
pub struct DeltaStackResult {
    pub boundaries: Vec<usize>,
    pub chunk_count: usize,
    pub thread_count: usize,
}

/// Core implementation of the Δ-Stack Monoid algorithm
///
/// This struct encapsulates the three-phase sentence boundary detection algorithm:
/// 1. Scan phase: Process chunks in parallel to compute partial states
/// 2. Prefix phase: Compute prefix sums to determine chunk start states
/// 3. Reduce phase: Combine partial results with start states to find boundaries
pub struct DeltaStackProcessor {
    language_rules: Arc<dyn LanguageRules>,
    chunk_manager: ChunkManager,
    parser: TextParser,
}

impl DeltaStackProcessor {
    /// Creates a new DeltaStackProcessor with the given configuration
    pub fn new(config: ProcessorConfig, language_rules: Arc<dyn LanguageRules>) -> Self {
        // Chunks must be strictly contiguous: overlapping chunks double-count
        // enclosure deltas in the prefix sum, which corrupts boundary
        // decisions for every chunk after the first overlap. The configured
        // overlap_size is therefore intentionally not forwarded here.
        let chunk_manager = ChunkManager::new(config.chunk_size, 0);

        Self {
            language_rules,
            chunk_manager,
            parser: TextParser::new(),
        }
    }

    /// Main processing method that executes the Δ-Stack Monoid algorithm
    pub fn process(&self, text: &str, mode: ExecutionMode) -> ProcessingResult<DeltaStackResult> {
        // Early return for empty text
        if text.is_empty() {
            return Ok(DeltaStackResult {
                boundaries: Vec::new(),
                chunk_count: 0,
                thread_count: 1,
            });
        }

        // Phase 0: Chunk the text
        let chunks = self.chunk_manager.chunk_text(text)?;
        if chunks.is_empty() {
            return Ok(DeltaStackResult {
                boundaries: Vec::new(),
                chunk_count: 0,
                thread_count: 1,
            });
        }

        let chunk_count = chunks.len();

        // Determine execution strategy. One thread pool is built per call and
        // shared by the scan and reduce phases (pool creation spawns OS
        // threads and is far too expensive to repeat per phase).
        let thread_count = mode.determine_thread_count(text.len());
        let pool = if thread_count > 1 {
            Some(
                rayon::ThreadPoolBuilder::new()
                    .num_threads(thread_count)
                    .build()
                    .map_err(|e| ProcessingError::InvalidConfig {
                        reason: format!("Failed to create thread pool: {e}"),
                    })?,
            )
        } else {
            None
        };

        // Execute the three phases
        let partial_states = self.scan_phase(&chunks, pool.as_ref());
        let chunk_starts = self.prefix_phase(&partial_states, &chunks)?;
        let boundaries = self.reduce_phase(&partial_states, &chunk_starts, &chunks, pool.as_ref());

        Ok(DeltaStackResult {
            boundaries,
            chunk_count,
            thread_count,
        })
    }

    /// Phase 1: Scan - Process chunks to compute partial states
    fn scan_phase(
        &self,
        chunks: &[TextChunk],
        pool: Option<&rayon::ThreadPool>,
    ) -> Vec<PartialState> {
        // Line offset at the start of each chunk, carried across chunks so
        // that line-start suppression rules don't mistake every chunk start
        // for a line start.
        let line_offsets: Vec<usize> = (0..chunks.len())
            .map(|i| Self::line_offset_at_chunk_start(chunks, i))
            .collect();

        let scan = |chunk: &TextChunk, line_offset: usize| {
            self.parser.scan_chunk_with_context(
                &chunk.content,
                self.language_rules.as_ref(),
                line_offset,
            )
        };

        if let Some(pool) = pool {
            pool.install(|| {
                chunks
                    .par_iter()
                    .zip(line_offsets.par_iter())
                    .map(|(chunk, &line_offset)| scan(chunk, line_offset))
                    .collect()
            })
        } else {
            chunks
                .iter()
                .zip(line_offsets.iter())
                .map(|(chunk, &line_offset)| scan(chunk, line_offset))
                .collect()
        }
    }

    /// Number of characters between the last newline before the given chunk
    /// and the chunk start. Line offsets are only ever compared against a
    /// small threshold (currently 10) by suppression rules, so the backward
    /// scan is capped and the returned value saturates at that cap.
    fn line_offset_at_chunk_start(chunks: &[TextChunk], index: usize) -> usize {
        const CAP: usize = 64;
        let mut offset = 0usize;
        for chunk in chunks[..index].iter().rev() {
            for c in chunk.content.chars().rev() {
                if c == '\n' || offset >= CAP {
                    return offset;
                }
                offset += 1;
            }
        }
        offset
    }

    /// Phase 2: Prefix - Compute chunk start states using prefix sums
    fn prefix_phase(
        &self,
        partial_states: &[PartialState],
        chunks: &[TextChunk],
    ) -> ProcessingResult<Vec<ChunkStartState>> {
        if partial_states.len() > 1 {
            Ok(PrefixSumComputer::compute_prefix_sum_with_overlap(
                partial_states,
                chunks,
            ))
        } else {
            // Single chunk - create default start state
            Ok(vec![ChunkStartState {
                cumulative_deltas: DeltaVec::from_vec(vec![
                    DeltaEntry { net: 0, min: 0 };
                    partial_states[0].deltas.len()
                ]),
                global_offset: 0,
            }])
        }
    }

    /// Phase 3: Reduce - Combine partial results with start states to find boundaries
    fn reduce_phase(
        &self,
        partial_states: &[PartialState],
        chunk_starts: &[ChunkStartState],
        chunks: &[TextChunk],
        pool: Option<&rayon::ThreadPool>,
    ) -> Vec<usize> {
        // Create pairs of (state, start) for processing
        let state_start_pairs: Vec<_> = partial_states
            .iter()
            .zip(chunk_starts.iter())
            .zip(chunks.iter())
            .collect();

        // Symmetric enclosure types are evaluated by parity in the reduce
        // phase; compute the mask once for all chunks.
        let symmetric_types = self.language_rules.symmetric_enclosure_types();

        // Process chunks to find boundaries
        let chunk_boundaries: Vec<Vec<Boundary>> = if let Some(pool) = pool {
            pool.install(|| {
                state_start_pairs
                    .par_iter()
                    .map(|((state, start), chunk)| {
                        self.reduce_chunk(state, start, chunk, &symmetric_types)
                    })
                    .collect()
            })
        } else {
            // Sequential reduction
            state_start_pairs
                .iter()
                .map(|((state, start), chunk)| {
                    self.reduce_chunk(state, start, chunk, &symmetric_types)
                })
                .collect()
        };

        // Merge boundaries from all chunks
        self.merge_boundaries(chunk_boundaries)
    }

    /// Reduces a single chunk to find its boundaries
    fn reduce_chunk(
        &self,
        state: &PartialState,
        start: &ChunkStartState,
        chunk: &TextChunk,
        symmetric_types: &[bool],
    ) -> Vec<Boundary> {
        // For single chunk, use reduce_single
        // For multiple chunks, we need proper indexing
        if chunk.total_chunks == 1 {
            // This is the only chunk
            BoundaryReducer::reduce_single_with_symmetry(state, symmetric_types)
        } else {
            // Multi-chunk - need to handle properly with correct offsets.
            // Ownership rule (start, end]: a boundary offset is the position
            // *after* its terminator, so a chunk's own content can only yield
            // offsets in (start_offset, end_offset]. Using an inclusive upper
            // bound keeps a boundary sitting exactly at the end of the text,
            // which the previous `< end_offset` filter dropped.
            BoundaryReducer::evaluate_candidates_with_symmetry(
                &state.boundary_candidates,
                start,
                symmetric_types,
            )
            .into_iter()
            .filter(|b| b.offset > chunk.start_offset && b.offset <= chunk.end_offset)
            .collect()
        }
    }

    /// Merges boundaries from multiple chunks into a single sorted list
    fn merge_boundaries(&self, chunk_boundaries: Vec<Vec<Boundary>>) -> Vec<usize> {
        let mut all_boundaries = Vec::new();

        for boundaries in chunk_boundaries {
            for boundary in boundaries {
                all_boundaries.push(boundary.offset);
            }
        }

        // Sort and deduplicate
        all_boundaries.sort_unstable();
        all_boundaries.dedup();

        all_boundaries
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::domain::language::ConfigurableLanguageRules;

    fn create_test_processor() -> DeltaStackProcessor {
        let config = ProcessorConfig::default();
        let rules = Arc::new(ConfigurableLanguageRules::from_code("en").unwrap());
        DeltaStackProcessor::new(config, rules)
    }

    #[test]
    fn test_empty_text() {
        let processor = create_test_processor();
        let result = processor.process("", ExecutionMode::Sequential).unwrap();
        assert!(result.boundaries.is_empty());
        assert_eq!(result.chunk_count, 0);
        assert_eq!(result.thread_count, 1);
    }

    #[test]
    fn test_single_sentence() {
        let processor = create_test_processor();
        let text = "This is a sentence.";
        let result = processor.process(text, ExecutionMode::Sequential).unwrap();
        assert_eq!(result.boundaries.len(), 1);
        assert_eq!(result.boundaries[0], 19); // Position after the period
        assert_eq!(result.chunk_count, 1);
        assert_eq!(result.thread_count, 1);
    }

    #[test]
    fn test_parallel_vs_sequential() {
        let processor = create_test_processor();
        let text = "First sentence. Second sentence. Third sentence.";

        let seq_result = processor.process(text, ExecutionMode::Sequential).unwrap();
        let par_result = processor
            .process(text, ExecutionMode::Parallel { threads: Some(2) })
            .unwrap();

        assert_eq!(seq_result.boundaries, par_result.boundaries);
        assert_eq!(seq_result.chunk_count, par_result.chunk_count);
        assert_eq!(seq_result.thread_count, 1);
        assert_eq!(par_result.thread_count, 2);
    }
}