use std::collections::HashMap;
use std::fs::File;
use std::io::{self, Read, ErrorKind}; // Import ErrorKind for better error handling
use std::thread;
// No external crates (like serde, rayon, etc.) are needed.

// --- Configuration ---
/// The path to the input text file.
const FILE_PATH: &str = "/tmp/file.txt";
/// The number of threads to use for parallel processing.
const NUM_THREADS: usize = 4;

/// Reads the file, splits the work among threads, counts words in parallel,
/// and prints one JSON output per thread.
fn main() -> io::Result<()> {
    // 1. Read the entire file content into a String
    // Provide more specific error messages if file operations fail.
    let mut file = File::open(FILE_PATH)
        .map_err(|e| io::Error::new(e.kind(), format!("Failed to open '{}': {}", FILE_PATH, e)))?;
    let mut content = String::new();
    file.read_to_string(&mut content)
        .map_err(|e| io::Error::new(e.kind(), format!("Failed to read '{}': {}", FILE_PATH, e)))?;

    // Handle empty file: Print N empty JSON objects immediately.
    if content.is_empty() {
        for _ in 0..NUM_THREADS {
            println!("{{}}");
        }
        return Ok(());
    }

    // 2. Determine chunk boundaries to divide the work among threads.
    // The goal is to find split points near (length / NUM_THREADS) intervals,
    // preferably at whitespace, to avoid splitting words across threads.
    let mut boundaries = vec![0]; // Stores the start index of each chunk's *potential* range.
    let total_len = content.len();

    if total_len > 0 { // Only calculate boundaries if there's content
        for i in 1..NUM_THREADS {
            // Calculate the approximate boundary point based on byte length.
            let approx_boundary = (total_len * i) / NUM_THREADS;

            // Find the index of the first whitespace character *at or after* the approximate boundary.
            // This ensures that the chunk processed by the *previous* thread ends cleanly before or at whitespace.
            // We search from the approximate boundary onwards.
            let boundary = content[approx_boundary..]
                .find(|c: char| c.is_whitespace())
                .map(|pos| approx_boundary + pos) // Calculate absolute index of the whitespace.
                .unwrap_or(total_len); // If no whitespace found until the end, the boundary is the end.

            boundaries.push(boundary.min(total_len)); // Add the boundary, ensuring it doesn't exceed total length.
        }
    }
    boundaries.push(total_len); // Add the final boundary (end of the string).

    // Sort and deduplicate boundaries. This handles cases where calculated boundaries
    // might be identical (e.g., multiple consecutive spaces at a boundary point).
    boundaries.sort_unstable();
    boundaries.dedup();

    // 3. Create (start, end) index pairs for the actual slices each thread will process.
    // We aim for exactly NUM_THREADS pairs, even if some are empty due to file size or boundary calculation.
    let mut chunks = Vec::with_capacity(NUM_THREADS);
    let mut last_boundary_end = 0;

    // Iterate through the calculated boundary points to define chunk ranges.
    // `boundaries` contains the end points (exclusive) of the ideal ranges.
    for i in 1..boundaries.len() {
        let current_boundary_end = boundaries[i];
        let start = last_boundary_end;
        let end = current_boundary_end;

        // Add the chunk definition (start, end). The thread will handle splitting words within this range.
        // Ensure start <= end, which should hold true after sort/dedup.
        chunks.push((start.min(end), end));

        last_boundary_end = current_boundary_end; // The end of this chunk is the start of the next.
    }

    // Ensure we have exactly NUM_THREADS chunk definitions. If boundary calculations
    // resulted in fewer actual chunks (e.g., very short file, or NUM_THREADS > number of words),
    // pad the `chunks` vector with empty ranges.
    while chunks.len() < NUM_THREADS {
        // Add an empty chunk range pointing to the end of the content.
        chunks.push((total_len, total_len));
    }
    // If, for some reason, we generated more chunks than threads (e.g., NUM_THREADS=1 but boundaries split), truncate.
    chunks.truncate(NUM_THREADS);


    // 4. Process chunks in parallel using `thread::scope`.
    // `thread::scope` ensures all spawned threads finish before `main` continues or returns.
    // This allows threads to safely borrow `content` without needing `Arc`.
    thread::scope(|s| {
        for i in 0..NUM_THREADS {
            // Get the specific (start, end) byte range for this thread.
            let (start, end) = chunks[i];

            // Borrow the content slice for this thread. Check bounds for safety, although
            // the calculation logic should ensure they are valid.
            if start <= end && end <= content.len() {
                // `content` is borrowed immutably here, which is safe for multiple threads.
                let chunk_str: &str = &content[start..end];

                // Spawn a new thread to process this chunk.
                s.spawn(move || {
                    // Each thread gets its own word count map.
                    let mut word_counts: HashMap<String, usize> = HashMap::new();

                    // `split_whitespace()` iterates over words separated by any amount of whitespace.
                    // It correctly handles leading/trailing whitespace within the `chunk_str`.
                    // It operates only on the provided slice, so words split exactly at the
                    // byte boundary `start` or `end` will be handled correctly by the respective threads.
                    for word in chunk_str.split_whitespace() {
                        // Increment the count for the word. `to_string()` creates an owned String key.
                        *word_counts.entry(word.to_string()).or_insert(0) += 1;
                    }

                    // Format and print this thread's result as a JSON object on its own line.
                    print_json(&word_counts);
                });
            } else {
                // This case should ideally not be reached with correct boundary/chunk logic.
                // However, to strictly guarantee NUM_THREADS outputs as requested,
                // spawn a thread that prints empty JSON if the chunk indices seem invalid.
                s.spawn(|| {
                    println!("{{}}");
                });
            }
        }
    }); // `scope` automatically waits for all spawned threads to complete here.

    Ok(())
}

/// Formats a HashMap<String, usize> into a JSON object string and prints it to stdout, followed by a newline.
/// Keys within the JSON object are sorted alphabetically for deterministic output.
fn print_json(counts: &HashMap<String, usize>) {
    // Handle the case of an empty map (no words found in the chunk).
    if counts.is_empty() {
        println!("{{}}");
        return;
    }

    let mut items = Vec::with_capacity(counts.len());

    // Sort keys for consistent output order. This is good practice, especially for testing.
    let mut sorted_keys: Vec<&String> = counts.keys().collect();
    sorted_keys.sort_unstable(); // `sort_unstable` might offer slightly better performance than stable sort.

    // Build the JSON parts: "key": value
    for key in sorted_keys {
        // We expect the key to be present, but `get` is safer than direct indexing.
        if let Some(count) = counts.get(key) {
            // Basic JSON string escaping for the key: only handles literal double quotes.
            // Words from split_whitespace typically don't contain quotes, but this adds robustness.
            let escaped_key = key.replace('"', "\\\"");
            items.push(format!("\"{}\": {}", escaped_key, count));
        }
    }

    // Join the key-value pairs with commas and wrap in curly braces for valid JSON.
    println!("{{{}}}", items.join(", "));
}

/*
// How to Compile and Run:
// 1. Save the code above as `rust_word_count.rs` (or any other name).
// 2. Create the input file:
//    echo "This is is is, an example file.\nAnother line is here, with is again." > /tmp/file.txt
// 3. Compile the Rust code using the standard Rust compiler (no Cargo needed):
//    rustc rust_word_count.rs
// 4. Run the compiled executable:
//    ./rust_word_count

// Example Expected Output (The order of the lines and the exact distribution of words
//                          will depend on the file content, length, and thread scheduling,
//                          but there will be exactly 4 lines of JSON):
// {"is": 2}
// {"This": 1, "is,": 1}
// {"an": 1, "example": 1, "file.": 1}
// {"Another": 1, "again.": 1, "here,": 1, "is": 1, "line": 1, "with": 1}

// Note: The word "is" might appear in counts from different threads if it spans across
// the calculated chunk boundaries. The merging step (done externally) would sum these up.
// Punctuation attached to words (like "is," or "file.") is treated as part of the word
// by `split_whitespace`.
*/

use std::collections::HashMap;
use std::fs::File;
use std::io::{self, Read, ErrorKind}; // Import ErrorKind for better error handling
use std::thread;
// No external crates (like serde, rayon, etc.) are needed.

// --- Configuration ---
/// The path to the input text file.
const FILE_PATH: &str = "/tmp/file.txt";
/// The number of threads to use for parallel processing.
const NUM_THREADS: usize = 4;

/// Reads the file, splits the work among threads, counts words in parallel,
/// and prints one JSON output per thread.
fn main() -> io::Result<()> {
    // 1. Read the entire file content into a String
    // Provide more specific error messages if file operations fail.
    let mut file = File::open(FILE_PATH)
        .map_err(|e| io::Error::new(e.kind(), format!("Failed to open '{}': {}", FILE_PATH, e)))?;
    let mut content = String::new();
    file.read_to_string(&mut content)
        .map_err(|e| io::Error::new(e.kind(), format!("Failed to read '{}': {}", FILE_PATH, e)))?;

    // Handle empty file: Print N empty JSON objects immediately.
    if content.is_empty() {
        for _ in 0..NUM_THREADS {
            println!("{{}}");
        }
        return Ok(());
    }

    // 2. Determine chunk boundaries to divide the work among threads.
    // The goal is to find split points near (length / NUM_THREADS) intervals,
    // preferably at whitespace, to avoid splitting words across threads.
    let mut boundaries = vec![0]; // Stores the start index of each chunk's *potential* range.
    let total_len = content.len();

    if total_len > 0 { // Only calculate boundaries if there's content
        for i in 1..NUM_THREADS {
            // Calculate the approximate boundary point based on byte length.
            let approx_boundary = (total_len * i) / NUM_THREADS;

            // Find the index of the first whitespace character *at or after* the approximate boundary.
            // This ensures that the chunk processed by the *previous* thread ends cleanly before or at whitespace.
            // We search from the approximate boundary onwards.
            let boundary = content[approx_boundary..]
                .find(|c: char| c.is_whitespace())
                .map(|pos| approx_boundary + pos) // Calculate absolute index of the whitespace.
                .unwrap_or(total_len); // If no whitespace found until the end, the boundary is the end.

            boundaries.push(boundary.min(total_len)); // Add the boundary, ensuring it doesn't exceed total length.
        }
    }
    boundaries.push(total_len); // Add the final boundary (end of the string).

    // Sort and deduplicate boundaries. This handles cases where calculated boundaries
    // might be identical (e.g., multiple consecutive spaces at a boundary point).
    boundaries.sort_unstable();
    boundaries.dedup();

    // 3. Create (start, end) index pairs for the actual slices each thread will process.
    // We aim for exactly NUM_THREADS pairs, even if some are empty due to file size or boundary calculation.
    let mut chunks = Vec::with_capacity(NUM_THREADS);
    let mut last_boundary_end = 0;

    // Iterate through the calculated boundary points to define chunk ranges.
    // `boundaries` contains the end points (exclusive) of the ideal ranges.
    for i in 1..boundaries.len() {
        let current_boundary_end = boundaries[i];
        let start = last_boundary_end;
        let end = current_boundary_end;

        // Add the chunk definition (start, end). The thread will handle splitting words within this range.
        // Ensure start <= end, which should hold true after sort/dedup.
        chunks.push((start.min(end), end));

        last_boundary_end = current_boundary_end; // The end of this chunk is the start of the next.
    }

    // Ensure we have exactly NUM_THREADS chunk definitions. If boundary calculations
    // resulted in fewer actual chunks (e.g., very short file, or NUM_THREADS > number of words),
    // pad the `chunks` vector with empty ranges.
    while chunks.len() < NUM_THREADS {
        // Add an empty chunk range pointing to the end of the content.
        chunks.push((total_len, total_len));
    }
    // If, for some reason, we generated more chunks than threads (e.g., NUM_THREADS=1 but boundaries split), truncate.
    chunks.truncate(NUM_THREADS);


    // 4. Process chunks in parallel using `thread::scope`.
    // `thread::scope` ensures all spawned threads finish before `main` continues or returns.
    // This allows threads to safely borrow `content` without needing `Arc`.
    thread::scope(|s| {
        for i in 0..NUM_THREADS {
            // Get the specific (start, end) byte range for this thread.
            let (start, end) = chunks[i];

            // Borrow the content slice for this thread. Check bounds for safety, although
            // the calculation logic should ensure they are valid.
            if start <= end && end <= content.len() {
                // `content` is borrowed immutably here, which is safe for multiple threads.
                let chunk_str: &str = &content[start..end];

                // Spawn a new thread to process this chunk.
                s.spawn(move || {
                    // Each thread gets its own word count map.
                    let mut word_counts: HashMap<String, usize> = HashMap::new();

                    // `split_whitespace()` iterates over words separated by any amount of whitespace.
                    // It correctly handles leading/trailing whitespace within the `chunk_str`.
                    // It operates only on the provided slice, so words split exactly at the
                    // byte boundary `start` or `end` will be handled correctly by the respective threads.
                    for word in chunk_str.split_whitespace() {
                        // Increment the count for the word. `to_string()` creates an owned String key.
                        *word_counts.entry(word.to_string()).or_insert(0) += 1;
                    }

                    // Format and print this thread's result as a JSON object on its own line.
                    print_json(&word_counts);
                });
            } else {
                // This case should ideally not be reached with correct boundary/chunk logic.
                // However, to strictly guarantee NUM_THREADS outputs as requested,
                // spawn a thread that prints empty JSON if the chunk indices seem invalid.
                s.spawn(|| {
                    println!("{{}}");
                });
            }
        }
    }); // `scope` automatically waits for all spawned threads to complete here.

    Ok(())
}

/// Formats a HashMap<String, usize> into a JSON object string and prints it to stdout, followed by a newline.
/// Keys within the JSON object are sorted alphabetically for deterministic output.
fn print_json(counts: &HashMap<String, usize>) {
    // Handle the case of an empty map (no words found in the chunk).
    if counts.is_empty() {
        println!("{{}}");
        return;
    }

    let mut items = Vec::with_capacity(counts.len());

    // Sort keys for consistent output order. This is good practice, especially for testing.
    let mut sorted_keys: Vec<&String> = counts.keys().collect();
    sorted_keys.sort_unstable(); // `sort_unstable` might offer slightly better performance than stable sort.

    // Build the JSON parts: "key": value
    for key in sorted_keys {
        // We expect the key to be present, but `get` is safer than direct indexing.
        if let Some(count) = counts.get(key) {
            // Basic JSON string escaping for the key: only handles literal double quotes.
            // Words from split_whitespace typically don't contain quotes, but this adds robustness.
            let escaped_key = key.replace('"', "\\\"");
            items.push(format!("\"{}\": {}", escaped_key, count));
        }
    }

    // Join the key-value pairs with commas and wrap in curly braces for valid JSON.
    println!("{{{}}}", items.join(", "));
}

/*
// How to Compile and Run:
// 1. Save the code above as `rust_word_count.rs` (or any other name).
// 2. Create the input file:
//    echo "This is is is, an example file.\nAnother line is here, with is again." > /tmp/file.txt
// 3. Compile the Rust code using the standard Rust compiler (no Cargo needed):
//    rustc rust_word_count.rs
// 4. Run the compiled executable:
//    ./rust_word_count

// Example Expected Output (The order of the lines and the exact distribution of words
//                          will depend on the file content, length, and thread scheduling,
//                          but there will be exactly 4 lines of JSON):
// {"is": 2}
// {"This": 1, "is,": 1}
// {"an": 1, "example": 1, "file.": 1}
// {"Another": 1, "again.": 1, "here,": 1, "is": 1, "line": 1, "with": 1}

// Note: The word "is" might appear in counts from different threads if it spans across
// the calculated chunk boundaries. The merging step (done externally) would sum these up.
// Punctuation attached to words (like "is," or "file.") is treated as part of the word
// by `split_whitespace`.
*/

use std::collections::HashMap;
use std::fs::File;
use std::io::{self, Read, ErrorKind}; // Import ErrorKind for better error handling
use std::thread;
// No external crates (like serde, rayon, etc.) are needed.

// --- Configuration ---
/// The path to the input text file.
const FILE_PATH: &str = "/tmp/file.txt";
/// The number of threads to use for parallel processing.
const NUM_THREADS: usize = 4;

/// Reads the file, splits the work among threads, counts words in parallel,
/// and prints one JSON output per thread.
fn main() -> io::Result<()> {
    // 1. Read the entire file content into a String
    // Provide more specific error messages if file operations fail.
    let mut file = File::open(FILE_PATH)
        .map_err(|e| io::Error::new(e.kind(), format!("Failed to open '{}': {}", FILE_PATH, e)))?;
    let mut content = String::new();
    file.read_to_string(&mut content)
        .map_err(|e| io::Error::new(e.kind(), format!("Failed to read '{}': {}", FILE_PATH, e)))?;

    // Handle empty file: Print N empty JSON objects immediately.
    if content.is_empty() {
        for _ in 0..NUM_THREADS {
            println!("{{}}");
        }
        return Ok(());
    }

    // 2. Determine chunk boundaries to divide the work among threads.
    // The goal is to find split points near (length / NUM_THREADS) intervals,
    // preferably at whitespace, to avoid splitting words across threads.
    let mut boundaries = vec![0]; // Stores the start index of each chunk's *potential* range.
    let total_len = content.len();

    if total_len > 0 { // Only calculate boundaries if there's content
        for i in 1..NUM_THREADS {
            // Calculate the approximate boundary point based on byte length.
            let approx_boundary = (total_len * i) / NUM_THREADS;

            // Find the index of the first whitespace character *at or after* the approximate boundary.
            // This ensures that the chunk processed by the *previous* thread ends cleanly before or at whitespace.
            // We search from the approximate boundary onwards.
            let boundary = content[approx_boundary..]
                .find(|c: char| c.is_whitespace())
                .map(|pos| approx_boundary + pos) // Calculate absolute index of the whitespace.
                .unwrap_or(total_len); // If no whitespace found until the end, the boundary is the end.

            boundaries.push(boundary.min(total_len)); // Add the boundary, ensuring it doesn't exceed total length.
        }
    }
    boundaries.push(total_len); // Add the final boundary (end of the string).

    // Sort and deduplicate boundaries. This handles cases where calculated boundaries
    // might be identical (e.g., multiple consecutive spaces at a boundary point).
    boundaries.sort_unstable();
    boundaries.dedup();

    // 3. Create (start, end) index pairs for the actual slices each thread will process.
    // We aim for exactly NUM_THREADS pairs, even if some are empty due to file size or boundary calculation.
    let mut chunks = Vec::with_capacity(NUM_THREADS);
    let mut last_boundary_end = 0;

    // Iterate through the calculated boundary points to define chunk ranges.
    // `boundaries` contains the end points (exclusive) of the ideal ranges.
    for i in 1..boundaries.len() {
        let current_boundary_end = boundaries[i];
        let start = last_boundary_end;
        let end = current_boundary_end;

        // Add the chunk definition (start, end). The thread will handle splitting words within this range.
        // Ensure start <= end, which should hold true after sort/dedup.
        chunks.push((start.min(end), end));

        last_boundary_end = current_boundary_end; // The end of this chunk is the start of the next.
    }

    // Ensure we have exactly NUM_THREADS chunk definitions. If boundary calculations
    // resulted in fewer actual chunks (e.g., very short file, or NUM_THREADS > number of words),
    // pad the `chunks` vector with empty ranges.
    while chunks.len() < NUM_THREADS {
        // Add an empty chunk range pointing to the end of the content.
        chunks.push((total_len, total_len));
    }
    // If, for some reason, we generated more chunks than threads (e.g., NUM_THREADS=1 but boundaries split), truncate.
    chunks.truncate(NUM_THREADS);


    // 4. Process chunks in parallel using `thread::scope`.
    // `thread::scope` ensures all spawned threads finish before `main` continues or returns.
    // This allows threads to safely borrow `content` without needing `Arc`.
    thread::scope(|s| {
        for i in 0..NUM_THREADS {
            // Get the specific (start, end) byte range for this thread.
            let (start, end) = chunks[i];

            // Borrow the content slice for this thread. Check bounds for safety, although
            // the calculation logic should ensure they are valid.
            if start <= end && end <= content.len() {
                // `content` is borrowed immutably here, which is safe for multiple threads.
                let chunk_str: &str = &content[start..end];

                // Spawn a new thread to process this chunk.
                s.spawn(move || {
                    // Each thread gets its own word count map.
                    let mut word_counts: HashMap<String, usize> = HashMap::new();

                    // `split_whitespace()` iterates over words separated by any amount of whitespace.
                    // It correctly handles leading/trailing whitespace within the `chunk_str`.
                    // It operates only on the provided slice, so words split exactly at the
                    // byte boundary `start` or `end` will be handled correctly by the respective threads.
                    for word in chunk_str.split_whitespace() {
                        // Increment the count for the word. `to_string()` creates an owned String key.
                        *word_counts.entry(word.to_string()).or_insert(0) += 1;
                    }

                    // Format and print this thread's result as a JSON object on its own line.
                    print_json(&word_counts);
                });
            } else {
                // This case should ideally not be reached with correct boundary/chunk logic.
                // However, to strictly guarantee NUM_THREADS outputs as requested,
                // spawn a thread that prints empty JSON if the chunk indices seem invalid.
                s.spawn(|| {
                    println!("{{}}");
                });
            }
        }
    }); // `scope` automatically waits for all spawned threads to complete here.

    Ok(())
}

/// Formats a HashMap<String, usize> into a JSON object string and prints it to stdout, followed by a newline.
/// Keys within the JSON object are sorted alphabetically for deterministic output.
fn print_json(counts: &HashMap<String, usize>) {
    // Handle the case of an empty map (no words found in the chunk).
    if counts.is_empty() {
        println!("{{}}");
        return;
    }

    let mut items = Vec::with_capacity(counts.len());

    // Sort keys for consistent output order. This is good practice, especially for testing.
    let mut sorted_keys: Vec<&String> = counts.keys().collect();
    sorted_keys.sort_unstable(); // `sort_unstable` might offer slightly better performance than stable sort.

    // Build the JSON parts: "key": value
    for key in sorted_keys {
        // We expect the key to be present, but `get` is safer than direct indexing.
        if let Some(count) = counts.get(key) {
            // Basic JSON string escaping for the key: only handles literal double quotes.
            // Words from split_whitespace typically don't contain quotes, but this adds robustness.
            let escaped_key = key.replace('"', "\\\"");
            items.push(format!("\"{}\": {}", escaped_key, count));
        }
    }

    // Join the key-value pairs with commas and wrap in curly braces for valid JSON.
    println!("{{{}}}", items.join(", "));
}

/*
// How to Compile and Run:
// 1. Save the code above as `rust_word_count.rs` (or any other name).
// 2. Create the input file:
//    echo "This is is is, an example file.\nAnother line is here, with is again." > /tmp/file.txt
// 3. Compile the Rust code using the standard Rust compiler (no Cargo needed):
//    rustc rust_word_count.rs
// 4. Run the compiled executable:
//    ./rust_word_count

// Example Expected Output (The order of the lines and the exact distribution of words
//                          will depend on the file content, length, and thread scheduling,
//                          but there will be exactly 4 lines of JSON):
// {"is": 2}
// {"This": 1, "is,": 1}
// {"an": 1, "example": 1, "file.": 1}
// {"Another": 1, "again.": 1, "here,": 1, "is": 1, "line": 1, "with": 1}

// Note: The word "is" might appear in counts from different threads if it spans across
// the calculated chunk boundaries. The merging step (done externally) would sum these up.
// Punctuation attached to words (like "is," or "file.") is treated as part of the word
// by `split_whitespace`.
*/